package com.sh.data.engine.domain.datadev.offline.service.impl;

import cn.hutool.core.convert.Convert;
import com.alibaba.druid.DbType;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.sh.data.engine.common.constants.Constants.RedisKey;
import com.sh.data.engine.common.enumDefinition.DSType;
import com.sh.data.engine.common.enumDefinition.DataQualityRuleCompareTypeEnum;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.ScheduleCycleUtil;
import com.sh.data.engine.common.util.ServletUtils;
import com.sh.data.engine.domain.base.model.PageResult;
import com.sh.data.engine.domain.base.model.UserContext;
import com.sh.data.engine.domain.base.model.enums.XxlGroupEnum;
import com.sh.data.engine.domain.datadev.offline.model.domain.*;
import com.sh.data.engine.domain.datadev.offline.model.domain.OfflineDevSyncDomain.FieldMapping;
import com.sh.data.engine.domain.datadev.offline.model.domain.OfflineDevSyncDomain.PartitionField;
import com.sh.data.engine.domain.datadev.offline.model.domain.OfflineDevToolsDomain.Node;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevApiNodeTypeEnum;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevNodeTypeEnum;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevTopNodeIdEnum;
import com.sh.data.engine.domain.datadev.offline.model.param.*;
import com.sh.data.engine.domain.datadev.offline.service.*;
import com.sh.data.engine.domain.dataquality.model.domain.DataQualityRuleTemplateDomain;
import com.sh.data.engine.domain.dataquality.service.DataQualityManagerService;
import com.sh.data.engine.domain.integration.api.http.model.domain.TaskDomain;
import com.sh.data.engine.domain.integration.api.http.model.param.ApiHttpRequestBodyParam;
import com.sh.data.engine.domain.integration.api.http.model.param.TestApiHttpParam;
import com.sh.data.engine.domain.integration.api.taskconfig.model.domain.TaskConfigDomain;
import com.sh.data.engine.domain.integration.api.webservice.model.param.ApiWsRequestBodyParam;
import com.sh.data.engine.domain.integration.api.webservice.model.param.TestApiWsParam;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceDetailDomain;
import com.sh.data.engine.domain.integration.datasource.service.DataSourceService;
import com.sh.data.engine.domain.integration.offline.model.domain.OfflineSyncDomain;
import com.sh.data.engine.domain.integration.offline.model.domain.OfflineTaskExistTableMappingDomain;
import com.sh.data.engine.domain.integration.offline.model.domain.OfflineTaskExistTableMappingDomain.Field;
import com.sh.data.engine.domain.integration.offline.model.domain.OfflineTaskExistTableMappingDomain.Table;
import com.sh.data.engine.domain.integration.offline.model.domain.OfflineTaskExistTableMappingQueryDomain;
import com.sh.data.engine.domain.integration.offline.service.OfflineSyncService;
import com.sh.data.engine.domain.secret.service.SecretMgtService;
import com.sh.data.engine.domain.shims.hdfs.util.HdfsUtil;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.domain.util.SqlParserUtil;
import com.sh.data.engine.domain.util.VarParserUtil;
import com.sh.data.engine.domain.workspace.index.service.ProjectService;
import com.sh.data.engine.domain.workspace.manager.model.domain.UserDomain;
import com.sh.data.engine.domain.workspace.manager.service.StorageService;
import com.sh.data.engine.domain.workspace.manager.service.UserService;
import com.sh.data.engine.infrastructure.config.HDFSConfig;
import com.sh.data.engine.job.admin.core.model.XxlJobInfo;
import com.sh.data.engine.job.admin.core.route.ExecutorRouteStrategyEnum;
import com.sh.data.engine.job.admin.core.scheduler.MisfireStrategyEnum;
import com.sh.data.engine.job.admin.core.scheduler.ScheduleTypeEnum;
import com.sh.data.engine.job.admin.service.XxlJobService;
import com.sh.data.engine.job.core.biz.model.ReturnT;
import com.sh.data.engine.job.core.enums.ExecutorBlockStrategyEnum;
import com.sh.data.engine.job.core.glue.GlueTypeEnum;
import com.sh.data.engine.repository.dao.datadev.entity.*;
import com.sh.data.engine.repository.dao.workspace.index.entity.ProjectEntity;
import com.sh.data.engine.repository.dao.workspace.manager.entity.StorageEntity;
import gudusoft.gsqlparser.EDbVendor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.io.File;
import java.math.BigDecimal;
import java.net.URI;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @Author: zigui.zdf @Date: 2022/2/21 11:14:31 @Description:
 */
@Service
@Slf4j
public class OfflineDevMenuNodeServiceImpl implements OfflineDevMenuNodeService {

    @Autowired
    private DatadevMenuNodeService datadevMenuNodeService;

    @Autowired
    private DatadevFunctionService datadevFunctionService;

    @Autowired
    private DatadevOfflineSyncService datadevOfflineSyncService;

    @Autowired
    private DatadevOfflineApiService datadevOfflineApiService;

    @Autowired
    private DatadevDataqualityService datadevDataqualityService;

    @Autowired
    private DatadevResourceService datadevResourceService;

    @Autowired
    private DatadevVnodeSerivce datadevVnodeSerivce;

    @Autowired
    private DatadevSubNodeDependencyService datadevSubNodeDependencyService;

    @Autowired
    private DatadevSubNodeService datadevSubNodeService;

    @Autowired
    private DatadevOperationHistoryService datadevOperationHistoryService;

    @Autowired
    private DatadevScriptService datadevScriptService;

    @Autowired
    private DatadevSyncPartitionService datadevSyncPartitionService;

    @Autowired
    private DatadevSyncFieldMappingService datadevSyncFieldMappingService;

    @Autowired
    private UserService userService;

    @Autowired
    private XxlJobService xxlJobService;

    @Autowired
    private DataSourceService dataSourceService;

    @Autowired
    private DatadevJobService datadevJobService;

    @Autowired
    private DataQualityManagerService dataQualityManagerService;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Lazy
    @Resource
    private ProjectService projectService;

    @Resource
    private StorageService storageService;

    @Autowired
    private OfflineSyncService offlineSyncService;

    @Lazy
    @Autowired
    private SecretMgtService secretMgtService;

    @Autowired
    private HDFSConfig hdfsConfig;

    /**
     * type
     */
    private static final List<Integer> DEFAULT_SYSTEM_ID = Lists.newArrayList(1, 2, 3, 4);

    private static final List<Integer> DEFAULT_TASKFLOW_ID = Lists.newArrayList(2, 3, 4);

    private static List<Long> TOP_NODE_ID = Lists.newArrayList();

    private static Map<Integer, Integer> DIR_TYPE = Maps.newHashMap();

    private static int GP_TEXT = 1;

    static {
        TOP_NODE_ID =
            Arrays.stream(OfflineDevTopNodeIdEnum.VALUES)
                .map(e -> Long.valueOf(e.getId()))
                .collect(Collectors.toList());

        DIR_TYPE.put(
            OfflineDevNodeTypeEnum.DIR_DATADEV.getNodeType(),
            OfflineDevNodeTypeEnum.DIR_OF_DATADEV.getNodeType());
        DIR_TYPE.put(
            OfflineDevNodeTypeEnum.DIR_DATAINTEGRATION.getNodeType(),
            OfflineDevNodeTypeEnum.DIR_OF_SYNCTASK.getNodeType());
        DIR_TYPE.put(
            OfflineDevNodeTypeEnum.DIR_DATAQUALITY.getNodeType(),
            OfflineDevNodeTypeEnum.DIR_OF_DATAQUALITY.getNodeType());
        DIR_TYPE.put(
            OfflineDevNodeTypeEnum.DIR_OF_DATADEV.getNodeType(),
            OfflineDevNodeTypeEnum.DIR_OF_DATADEV.getNodeType());
        DIR_TYPE.put(
            OfflineDevNodeTypeEnum.DIR_OF_SYNCTASK.getNodeType(),
            OfflineDevNodeTypeEnum.DIR_OF_SYNCTASK.getNodeType());
        DIR_TYPE.put(
            OfflineDevNodeTypeEnum.DIR_OF_DATAQUALITY.getNodeType(),
            OfflineDevNodeTypeEnum.DIR_OF_DATAQUALITY.getNodeType());
        DIR_TYPE.put(
            OfflineDevNodeTypeEnum.DIR_NORMAL.getNodeType(),
            OfflineDevNodeTypeEnum.DIR_NORMAL.getNodeType());
    }

    @Override
    public List<OfflineDevNodeDomain> getAllNodeByProjectIdAndPids(
        List<Long> pidList, Long projectId) {
        List<DatadevMenuNodeEntity> allNodes = new ArrayList<>();

        final List<DatadevMenuNodeEntity> nodeEntities =
            datadevMenuNodeService.list(
                new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                    .eq(DatadevMenuNodeEntity::getProjectId, projectId)
                    .eq(DatadevMenuNodeEntity::getRowState, 1));

        final List<DatadevMenuNodeEntity> nodePid =
            datadevMenuNodeService.list(
                new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                    .eq(DatadevMenuNodeEntity::getProjectId, projectId)
                    .in(DatadevMenuNodeEntity::getPid, pidList)
                    .eq(DatadevMenuNodeEntity::getRowState, 1));

        // 先查询出所有的节点
        allNodes.addAll(nodePid);

        Map<Long, List<DatadevMenuNodeEntity>> pidNodesMap =
            nodeEntities.stream().collect(Collectors.groupingBy(DatadevMenuNodeEntity::getPid));

        for (DatadevMenuNodeEntity nodeEntity : nodePid) {
            final OfflineDevNodeTypeEnum nodeTypeEnum =
                OfflineDevNodeTypeEnum.of(nodeEntity.getNodeType());
            if (nodeTypeEnum.isDir()) {
                getAllNodeByPid(nodeEntity.getId(), allNodes, pidNodesMap);
            }
        }

        final List<OfflineDevNodeDomain> offlineDevNodeDomains =
            ConvertUtil.copyProperties(allNodes, OfflineDevNodeDomain.class);

        List<OfflineDevNodeDomain> offlineDevNodeDomainList =
            offlineDevNodeDomains.stream()
                .filter(
                    e ->
                        (null != e.getNodeType()
                            && !e.getNodeType()
                            .equals(OfflineDevNodeTypeEnum.VNODE_START.getNodeType())))
                .map(
                    e -> {
                        if (null == e.getSort()) {
                            e.setSort(0);
                        }
                        return e;
                    })
                .sorted(Comparator.comparing(OfflineDevNodeDomain::getSort))
                .collect(Collectors.toList());
        return offlineDevNodeDomainList;
    }

    @Override
    public List<OfflineDevNodeDomain> getAllNodeByProjectId(Long pid, Long projectId) {
        List<DatadevMenuNodeEntity> allNodes = new ArrayList<>();

        final List<DatadevMenuNodeEntity> nodeEntities =
            datadevMenuNodeService.list(
                new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                    .eq(DatadevMenuNodeEntity::getProjectId, projectId)
                    .eq(DatadevMenuNodeEntity::getRowState, 1));

        final List<DatadevMenuNodeEntity> nodePid =
            datadevMenuNodeService.list(
                new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                    .eq(DatadevMenuNodeEntity::getProjectId, projectId)
                    .eq(DatadevMenuNodeEntity::getPid, pid)
                    .eq(DatadevMenuNodeEntity::getRowState, 1));

        // 先查询出所有的节点
        allNodes.addAll(nodePid);

        Map<Long, List<DatadevMenuNodeEntity>> pidNodesMap =
            nodeEntities.stream().collect(Collectors.groupingBy(DatadevMenuNodeEntity::getPid));

        for (DatadevMenuNodeEntity nodeEntity : nodePid) {
            final OfflineDevNodeTypeEnum nodeTypeEnum =
                OfflineDevNodeTypeEnum.of(nodeEntity.getNodeType());
            if (nodeTypeEnum.isDir()) {
                getAllNodeByPid(nodeEntity.getId(), allNodes, pidNodesMap);
            }
        }

        final List<OfflineDevNodeDomain> offlineDevNodeDomains =
            ConvertUtil.copyProperties(allNodes, OfflineDevNodeDomain.class);

        List<OfflineDevNodeDomain> offlineDevNodeDomainList =
            offlineDevNodeDomains.stream()
                .filter(
                    e ->
                        (null != e.getNodeType()
                            && !e.getNodeType()
                            .equals(OfflineDevNodeTypeEnum.VNODE_START.getNodeType())))
                .map(
                    e -> {
                        if (null == e.getSort()) {
                            e.setSort(0);
                        }
                        return e;
                    })
                .sorted(Comparator.comparing(OfflineDevNodeDomain::getSort))
                .collect(Collectors.toList());
        return offlineDevNodeDomainList;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean deleteNode(
        Long id, Long projectId, boolean includeChildTask, boolean physicsDelete, Long taskFlowId) {
        final DatadevMenuNodeEntity entity = datadevMenuNodeService.getById(id);
        if (entity == null) {
            throw new BusinessException("节点不存在");
        }

        final OfflineDevNodeTypeEnum nodeTypeEnum = OfflineDevNodeTypeEnum.of(entity.getNodeType());

        if (null != entity.getIsSystem() && entity.getIsSystem().equals(1)) {
            throw new BusinessException("该节点不支持删除");
        }

        if (Objects.equals(nodeTypeEnum, OfflineDevNodeTypeEnum.VNODE_START)) {
            throw new BusinessException("该节点不支持删除");
        }

        if (nodeTypeEnum.isTaskFlowDefaultDir()) {
            throw new BusinessException("该节点不支持删除");
        }

        // 物理删除
        if (physicsDelete) {

            /** 删除连线和工作流关系 */

            // todo 判断节点是否在工作流内，如果是，删除相关引用
            if (nodeTypeEnum.isTask()) {
                deleteTask(id);
            }

            if (Objects.equals(nodeTypeEnum, OfflineDevNodeTypeEnum.RESOURCE)) {
                deleteResource(id);
            }

            if (Objects.equals(nodeTypeEnum, OfflineDevNodeTypeEnum.FUNCTION)) {
                deleteFunction(id);
            }

            if (Objects.equals(nodeTypeEnum, OfflineDevNodeTypeEnum.DIR_TASK_FLOW)) {
                deleteTaskFlow(id, includeChildTask);
            }

            if (Objects.equals(nodeTypeEnum, OfflineDevNodeTypeEnum.DIR_NORMAL)) {
                deleteDir(id);
            }

            return true;
        }

        if (Objects.nonNull(taskFlowId) && !taskFlowId.equals(id)) {

            final List<DatadevSubNodeEntity> subTaskList =
                datadevSubNodeService.list(
                    new LambdaQueryWrapper<DatadevSubNodeEntity>()
                        .eq(DatadevSubNodeEntity::getTaskFlowId, taskFlowId));
            final List<DatadevSubNodeDependencyEntity> subTaskDependencyList =
                datadevSubNodeDependencyService.list(
                    new LambdaQueryWrapper<DatadevSubNodeDependencyEntity>()
                        .eq(DatadevSubNodeDependencyEntity::getTaskFlowId, taskFlowId));

            if (CollectionUtils.isNotEmpty(subTaskDependencyList)) {
                Iterator<DatadevSubNodeDependencyEntity> iterator = subTaskDependencyList.iterator();
                while (iterator.hasNext()) {
                    DatadevSubNodeDependencyEntity next = iterator.next();
                    Long upNodeId = next.getUpNodeId();
                    Long downNodeId = next.getDownNodeId();
                    if ((Objects.nonNull(upNodeId) && upNodeId.equals(id))
                        || (Objects.nonNull(downNodeId) && downNodeId.equals(id))) {
                        next.setRowState(0);
                    }
                }
            }

            Iterator<DatadevSubNodeEntity> iterator1 = subTaskList.iterator();
            while (iterator1.hasNext()) {
                DatadevSubNodeEntity next = iterator1.next();
                Long nodeId = next.getNodeId();
                if (Objects.nonNull(nodeId) && nodeId.equals(id)) {
                    next.setRowState(0);
                }
            }
            String msg = String.format("删除%s节点:%s", nodeTypeEnum.getNodeDesc(), entity.getName());

            datadevOperationHistoryService.insertHistory(entity.getId(), projectId, msg);

            datadevSubNodeService.updateBatchById(subTaskList);

            datadevSubNodeDependencyService.updateBatchById(subTaskDependencyList);
        }

        if (Objects.equals(nodeTypeEnum, OfflineDevNodeTypeEnum.DIR_NORMAL)) {
            LambdaQueryWrapper<DatadevMenuNodeEntity> eq =
                new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                    .eq(DatadevMenuNodeEntity::getPid, id)
                    .eq(DatadevMenuNodeEntity::getRowState, 1);
            List<DatadevMenuNodeEntity> list = datadevMenuNodeService.list(eq);
            for (DatadevMenuNodeEntity datadevMenuNodeEntity : list) {
                deleteNode(
                    datadevMenuNodeEntity.getId(), projectId, includeChildTask, physicsDelete, taskFlowId);
            }
        }

        // 逻辑删除
        entity.setRowState(0);
        datadevMenuNodeService.updateById(entity);

        // 同步删除 运维记录
        datadevJobService.dropRecordsByNodeId(id);

//        try {
//            bizDataIndexService.deleteIndexEs(BizDataIndexTypeEnum.OFFLINE_JOB, projectId, id);
//            bizDataIndexService.deleteIndexEs(BizDataIndexTypeEnum.OFFLINE_TASKFLOW, projectId, id);
//        } catch (Exception e) {
//            log.error("删除es数据失败{}", e.getMessage());
//        }
////        LogContext.put(LogKey.NAME.getKey(), entity.getName());
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean restoreNode(Long id, Long projectId) {
        final DatadevMenuNodeEntity entity = datadevMenuNodeService.getById(id);
        if (entity == null) {
            throw new BusinessException("节点不存在");
        }

        entity.setRowState(1);
        datadevMenuNodeService.updateById(entity);
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean editNodeName(Long id, Long projectId, String name) {
        final DatadevMenuNodeEntity entity = datadevMenuNodeService.getById(id);
        if (entity == null) {
            throw new BusinessException("节点不存在");
        }

        final boolean hasSameNode =
            datadevMenuNodeService.hasSameNode(id, entity.getPid(), name, projectId);
        if (hasSameNode) {
            throw new BusinessException("该父节点下存在相同名称的节点");
        }
        entity.setName(name);
        datadevMenuNodeService.updateById(entity);
//        LogContext.put(LogKey.NAME.getKey(), entity.getName());
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean clearAllNode(Long projectId) {

        final List<DatadevMenuNodeEntity> entityList =
            datadevMenuNodeService.list(
                new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                    .eq(DatadevMenuNodeEntity::getRowState, 0)
                    .eq(DatadevMenuNodeEntity::getProjectId, projectId));

        for (DatadevMenuNodeEntity datadevMenuNodeEntity : entityList) {
            try {
                this.deleteNode(datadevMenuNodeEntity.getId(), projectId, true, true, null);
            } catch (Exception ignore) {

            }
        }

        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean changeDir(Long id, Long pid, Long projectId) {
        final DatadevMenuNodeEntity target = datadevMenuNodeService.getById(id);
        final DatadevMenuNodeEntity parent = datadevMenuNodeService.getById(pid);

        if (target == null || parent == null) {
            throw new BusinessException("节点不存在");
        }

        final OfflineDevNodeTypeEnum targetNodeType = OfflineDevNodeTypeEnum.of(target.getNodeType());
        final OfflineDevNodeTypeEnum parentNodeType = OfflineDevNodeTypeEnum.of(parent.getNodeType());

        // 如果父节点不是目录,则不支持拖动
        if (!parentNodeType.isDir()) {
            throw new BusinessException("不支持的操作");
        }

        if (targetNodeType.isTaskFlowDefaultDir()) {
            throw new BusinessException("不支持的操作");
        }

        if (Objects.equals(target.getPid(), pid)) {
            throw new BusinessException("已在当前目录中");
        }

        final DatadevMenuNodeEntity targetTopTaskFlowDefaultDir = this.getTopTaskFlowDefaultDir(id);
        final DatadevMenuNodeEntity parentTopTaskFlowDefaultDir = this.getTopTaskFlowDefaultDir(pid);

        // 若两者都不是null，则认为是工作流
        if (null != targetTopTaskFlowDefaultDir && null != parentTopTaskFlowDefaultDir) {
            // 若两者任务组件类型不一样，则不支持修改
            if (!Objects.equals(
                targetTopTaskFlowDefaultDir.getNodeType(), parentTopTaskFlowDefaultDir.getNodeType())) {
                throw new BusinessException("不支持的操作");
            }

            // 若两者不是同一工作流内的节点，则需要特殊处理
            if (!Objects.equals(
                targetTopTaskFlowDefaultDir.getId(), parentTopTaskFlowDefaultDir.getId())) {
                // 1.删除原工作流程中的连线
                // 2.修改该节点的父节点（下面的逻辑有处理）
                // 3.修改该节点到新的工作流程中

                if (targetNodeType.isNormalDir()) {
                    List<DatadevMenuNodeEntity> allNodes = new ArrayList<>();
                    getAllNodeByPid(target.getId(), allNodes, projectId);

                    for (DatadevMenuNodeEntity node : allNodes) {
                        if (OfflineDevNodeTypeEnum.of(node.getNodeType()).isTask()) {
                            this.setTaskNodeToNewTaskFlow(
                                node.getId(),
                                targetTopTaskFlowDefaultDir.getPid(),
                                parentTopTaskFlowDefaultDir.getPid());
                        }
                    }
                } else {
                    this.setTaskNodeToNewTaskFlow(
                        target.getId(),
                        targetTopTaskFlowDefaultDir.getPid(),
                        parentTopTaskFlowDefaultDir.getPid());
                }
            }
        }

        final boolean hasSameNode =
            datadevMenuNodeService.hasSameNode(id, pid, target.getName(), projectId);
        if (hasSameNode) {
            throw new BusinessException("已存在相同名称的节点");
        }

        final Integer maxSort = datadevMenuNodeService.getMaxSortByPid(pid);
        Integer sort = maxSort + 1;

        target.setSort(sort);
        target.setPid(pid);
        datadevMenuNodeService.updateById(target);

        String msg = String.format("新移入%s节点:%s", targetNodeType.getNodeDesc(), target.getName());

        datadevOperationHistoryService.insertHistory(target.getId(), projectId, msg);

        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean sortDir(Long id, Long siblingsId, Integer dropPosition, Long projectId) {
        // 1.节点类型不一致不能排序
        final DatadevMenuNodeEntity target = datadevMenuNodeService.getById(id);
        final DatadevMenuNodeEntity siblings = datadevMenuNodeService.getById(siblingsId);

        if (null == target || null == siblings) {
            throw new BusinessException("节点不存在");
        }

        final OfflineDevNodeTypeEnum targetNodeType = OfflineDevNodeTypeEnum.of(target.getNodeType());
        final OfflineDevNodeTypeEnum siblingsNodeType =
            OfflineDevNodeTypeEnum.of(siblings.getNodeType());

        if (targetNodeType.isTaskFlowDefaultDir() || siblingsNodeType.isTaskFlowDefaultDir()) {
            throw new BusinessException("不支持的操作");
        }

        final DatadevMenuNodeEntity targetTopTaskFlowDefaultDir =
            this.getTopTaskFlowDefaultDir(target.getPid());
        final DatadevMenuNodeEntity siblingsTopTaskFlowDefaultDir =
            this.getTopTaskFlowDefaultDir(siblings.getPid());

        // 如果两者都不为空，则认为是工作流
        if (null != targetTopTaskFlowDefaultDir && null != siblingsTopTaskFlowDefaultDir) {
            // 若两者任务组件类型不一样，则不支持修改
            if (!Objects.equals(
                targetTopTaskFlowDefaultDir.getNodeType(), siblingsTopTaskFlowDefaultDir.getNodeType())) {
                throw new BusinessException("不支持的操作");
            }

            if (!Objects.equals(
                targetTopTaskFlowDefaultDir.getId(), siblingsTopTaskFlowDefaultDir.getId())) {
                // 1.删除原工作流程中的连线
                // 2.修改该节点的父节点（下面的逻辑有处理）
                // 3.修改该节点到新的工作流程中

                if (targetNodeType.isNormalDir()) {
                    List<DatadevMenuNodeEntity> allNodes = new ArrayList<>();
                    getAllNodeByPid(target.getId(), allNodes, projectId);

                    for (DatadevMenuNodeEntity node : allNodes) {
                        if (OfflineDevNodeTypeEnum.of(node.getNodeType()).isTask()) {
                            this.setTaskNodeToNewTaskFlow(
                                node.getId(),
                                targetTopTaskFlowDefaultDir.getPid(),
                                siblingsTopTaskFlowDefaultDir.getPid());
                        }
                    }
                } else {
                    this.setTaskNodeToNewTaskFlow(
                        target.getId(),
                        targetTopTaskFlowDefaultDir.getPid(),
                        siblingsTopTaskFlowDefaultDir.getPid());
                }
            }
        }

        // 若目标父节点下存在相同名称的节点，则失败
        final boolean hasSameNode =
            datadevMenuNodeService.hasSameNode(id, siblings.getPid(), target.getName(), projectId);
        if (hasSameNode) {
            throw new BusinessException("已存在相同名称的节点");
        }

        target.setPid(siblings.getPid());

        Integer siblingsDirSort = siblings.getSort() == null ? 1 : siblings.getSort();

        if (dropPosition == -1) {
            target.setSort(siblingsDirSort);
            siblings.setSort(siblingsDirSort + 1);
        } else {
            target.setSort(siblingsDirSort + 1);
        }

        List<DatadevMenuNodeEntity> nodes = new ArrayList<>();
        this.getAllNodeByPid(siblings.getPid(), nodes, projectId);

        nodes.stream()
            .filter(node -> node.getSort().compareTo(siblingsDirSort) > 0)
            .forEach(
                node -> {
                    node.setSort(node.getSort() + 1);
                });

        datadevMenuNodeService.updateBatchById(nodes);
        datadevMenuNodeService.updateById(target);
        datadevMenuNodeService.updateById(siblings);

        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean saveDir(Long id, Long pid, String name, Long projectId, String userId) {
        DatadevMenuNodeEntity menuNode = new DatadevMenuNodeEntity();
        menuNode.setId(id);
        menuNode.setName(name);
        menuNode.setPid(pid);
        menuNode.setProjectId(projectId);
        menuNode.setIsSystem(0);
        menuNode.setRowState(1);
        menuNode.setUpdaterId(userId);
        if (TOP_NODE_ID.contains(pid)) {
            menuNode.setNodeType(OfflineDevNodeTypeEnum.DIR_NORMAL.getNodeType());
        } else {
            DatadevMenuNodeEntity menuNodeEntity = datadevMenuNodeService.getById(pid);
            menuNode.setNodeType(DIR_TYPE.get(menuNodeEntity.getNodeType()));
        }

        //    final OfflineDevMenuNodeEntity dir =
        //        offlineDevMenuNodeMapper.findByIdAndProjectId(pid, projectId);
        //    if ((dir == null || !OfflineDevNodeTypeEnum.of(dir.getNodeType()).isDir())) {
        //      throw new BusinessException("父节点不存在或者父节点不是目录");
        //    }

        final boolean hasSameNode = datadevMenuNodeService.hasSameNode(id, pid, name, projectId);
        if (hasSameNode) {
            throw new BusinessException("该父节点下存在相同名称的节点");
        }

        if (menuNode.getId() != null) {

            if (Objects.equals(menuNode.getId(), menuNode.getPid())) {
                throw new BusinessException("不能将目录放在自身之下");
            }
            datadevMenuNodeService.updateById(menuNode);

            return true;
        }

        final Integer maxSortByPid = datadevMenuNodeService.getMaxSortByPid(pid);

        Integer sort = maxSortByPid + 1;

        menuNode.setSort(sort);
        menuNode.setCreatorId(userId);

        datadevMenuNodeService.save(menuNode);
//        LogContext.put(LogKey.NAME.getKey(), menuNode.getName());
        return true;
    }

    @Override
    public List<OfflineDevDirTreeDomain> getDirTree(Long id, Long projectId) {
        List<OfflineDevDirTreeDomain> responses = new ArrayList<>();

        /**
         * 1.先找到topNode topNode有以下几种情况 (nodeType) a. 1, "工作流程目录"(useless) b. 2, "数据集成目录" c.3, "数据开发目录"
         * d.4, "数据质量目录" (pid) "1", "作业开发" "2", "工作流程" "3", "资源管理" "4", "函数管理" "5", "任务运维"
         *
         * <p>2.从上到下递归找到所有的dir
         *
         * <p>3.组装成tree
         */
        DatadevMenuNodeEntity topNode = getTopNode(id);

        List<DatadevMenuNodeEntity> nodes = new ArrayList<>();

        // 如果是外部的寻找所有的
        if (TOP_NODE_ID.contains(topNode.getPid())) {
            getAllNodeIncludeByPid(topNode.getPid(), nodes, projectId);
            OfflineDevDirTreeDomain domain = new OfflineDevDirTreeDomain();
            OfflineDevTopNodeIdEnum topNodeIdEnum =
                OfflineDevTopNodeIdEnum.of(String.valueOf(topNode.getPid()));
            domain.setId(Long.valueOf(topNodeIdEnum.getId()));
            domain.setName(topNodeIdEnum.getName());
            getChildDirs(topNode.getPid(), nodes, domain);
            responses.add(domain);
        } else if (DEFAULT_SYSTEM_ID.contains(topNode.getNodeType())) {
            getAllSystemNodeByPid(topNode.getPid(), nodes, projectId);
            OfflineDevDirTreeDomain domain = new OfflineDevDirTreeDomain();
            domain.setId(topNode.getId());
            domain.setName(topNode.getName());
            domain.setPid(topNode.getPid());
            domain.setNodeType(topNode.getNodeType());
            domain.setSort(topNode.getSort());
            getChildDirs(topNode.getId(), nodes, domain);
            responses.add(domain);
        }

        //    if (OfflineDevTopNodeIdEnum.TASK_FLOW.getId().equals(topNode.getId())) {
        //      final OfflineDevMenuNodeEntity topTaskFlowDefaultDir = getTopTaskFlowDefaultDir(id);
        //      if (null != topTaskFlowDefaultDir) {
        //        topNode = topTaskFlowDefaultDir;
        //        getAllNodeByPid(topNode.getId(), nodes);
        //      } else {
        //        getTaskFlowDirNodes(OfflineDevTopNodeIdEnum.TASK_FLOW.getId(), nodes);
        //      }
        //    } else {
        //      getAllNodeByPid(topNode.getId(), nodes);
        //    }

        //    OfflineDevDirTreeDomain domain = new OfflineDevDirTreeDomain();
        //    domain.setId(topNode.getId());
        //    domain.setName(topNode.getName());
        //    domain.setPid(topNode.getPid());
        //    domain.setNodeType(topNode.getNodeType());
        //    getChildDirs(topNode.getId(), nodes, domain);
        //
        //    responses.add(domain);

        List<OfflineDevDirTreeDomain> r1 =
            responses.stream()
                .filter(e -> Objects.nonNull(e))
                .map(
                    e -> {
                        if (null == e.getSort()) {
                            e.setSort(0);
                        }
                        return e;
                    })
                .sorted(Comparator.comparing(OfflineDevDirTreeDomain::getSort))
                .collect(Collectors.toList());
        return r1;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean saveTask(OfflineDevTaskSaveParam param, Long projectId, String userId) {
        DatadevMenuNodeEntity entity = new DatadevMenuNodeEntity();
        entity.setId(param.getId());
        entity.setPid(param.getPid());
        entity.setApiType(param.getApiType());
        entity.setName(param.getNodeName());
        entity.setRemark(param.getRemark());
        entity.setUpdaterId(userId);

        Integer maxSort = datadevMenuNodeService.getMaxSortByPid(param.getPid());
        maxSort = (null == maxSort) ? 0 : maxSort;
        // 修改
        if (null != param.getId()) {
            final DatadevMenuNodeEntity nodeEntity = datadevMenuNodeService.getById(param.getId());
            if (null == nodeEntity) {
                throw new BusinessException("节点不存在");
            }

            final boolean hasSameNode =
                datadevMenuNodeService.hasSameNode(
                    param.getId(), param.getPid(), param.getNodeName(), projectId);
            if (hasSameNode) {
                throw new BusinessException("同名节点已存在");
            }
            nodeEntity.setPid(param.getPid());
            nodeEntity.setRemark(param.getRemark());
            nodeEntity.setName(param.getNodeName());
            nodeEntity.setApiType(param.getApiType());
            nodeEntity.setSort(maxSort + 1);
            nodeEntity.setUpdaterId(userId);
            datadevMenuNodeService.updateById(nodeEntity);
            //        LogContext.put(LogKey.NAME.getKey(), param.getNodeName());
            //        LogContext.put(LogKey.ISUPDATE.getKey(), LogKey.ISUPDATE.getValue());
            datadevOperationHistoryService.insertHistory(
                param.getId(), projectId, String.format("修改任务节点:[%s]的信息", param.getNodeName()));

            return true;
        }

        final boolean hasSameNode =
            datadevMenuNodeService.hasSameNode(null, param.getPid(), param.getNodeName(), projectId);
        if (hasSameNode) {
            throw new BusinessException("同名节点已存在");
        }

        entity.setCreatorId(userId);
        entity.setProjectId(projectId);
        entity.setNodeType(param.getNodeType());
        entity.setApiType(param.getApiType());
        entity.setRowState(1);
        entity.setSort(maxSort + 1);

        datadevMenuNodeService.save(entity);

        // 若是工作流，还得将任务节点放置到工作流内
        final DatadevMenuNodeEntity topTaskFlowDefaultDir = getTopTaskFlowDefaultDir(param.getPid());
        if (null != topTaskFlowDefaultDir) {

            List<DatadevSubNodeDependencyEntity> subTaskListDepency =
                datadevSubNodeDependencyService.list(
                    new LambdaQueryWrapper<DatadevSubNodeDependencyEntity>()
                        .eq(
                            DatadevSubNodeDependencyEntity::getTaskFlowId,
                            topTaskFlowDefaultDir.getPid()));
            List<DatadevSubNodeEntity> subTaskList =
                datadevSubNodeService.list(
                    new LambdaQueryWrapper<DatadevSubNodeEntity>()
                        .eq(DatadevSubNodeEntity::getTaskFlowId, topTaskFlowDefaultDir.getPid()));

            final String maxCoordX =
                subTaskList.stream()
                    .max(Comparator.comparing(st -> new BigDecimal(st.getCoordX())))
                    .get()
                    .getCoordX();
            final String maxCoordY =
                subTaskList.stream()
                    .max(Comparator.comparing(st -> new BigDecimal(st.getCoordY())))
                    .get()
                    .getCoordY();

            DatadevSubNodeEntity subTask = new DatadevSubNodeEntity();
            subTask.setNodeId(entity.getId());
            subTask.setRowState(1);
            subTask.setCoordX(new BigDecimal(maxCoordX).add(new BigDecimal("50")).toString());
            subTask.setCoordY(new BigDecimal(maxCoordY).add(new BigDecimal("50")).toString());
            subTask.setTaskFlowId(topTaskFlowDefaultDir.getPid());
            datadevSubNodeService.save(subTask);

            final OfflineDevNodeTypeEnum nodeTypeEnum = OfflineDevNodeTypeEnum.of(param.getNodeType());
            String msg = String.format("新增%s节点:%s", nodeTypeEnum.getNodeDesc(), param.getNodeName());

            datadevOperationHistoryService.insertHistory(entity.getId(), projectId, msg);
        }
//        LogContext.put(LogKey.NAME.getKey(), param.getNodeName());

        return true;
    }

    private boolean parentIsTopParent(String pid) {
        OfflineDevTopNodeIdEnum[] values = OfflineDevTopNodeIdEnum.VALUES;
        for (OfflineDevTopNodeIdEnum topNodeIdEnum : values) {
            if (topNodeIdEnum.getId().equalsIgnoreCase(pid)) {
                return true;
            }
        }
        return false;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean saveResourceTask(
        OfflineDevTaskResourceSaveParam param, Long projectId, String userId) {

        final Integer maxSort = datadevMenuNodeService.getMaxSortByPid(param.getPid());
        UserDomain userDomain = userService.getUserById(userId);
        // 修改
        if (null != param.getNodeId()) {
            final DatadevMenuNodeEntity nodeEntity = datadevMenuNodeService.getById(param.getNodeId());
            if (null == nodeEntity) {
                throw new BusinessException("节点不存在");
            }

            List<DatadevResourceEntity> resourceConfig =
                datadevResourceService.getResourceByNodeId(param.getNodeId());
            if (StringUtils.isNotBlank(param.getFilePath())
                && StringUtils.isNotBlank(param.getFilePath())
                && StringUtils.isNotBlank(param.getOriginalFilename())) {
                resourceConfig.forEach(
                    e -> {
                        if (e.getFileName().equals(param.getOriginalFilename())) {
                            throw new BusinessException("请更换文件名称，该文件已经存在");
                        }
                        e.setIsActive(0);
                    });

                // 将本地文件上传到hdfs

                DatadevResourceEntity resource = new DatadevResourceEntity();
                resource.setIsActive(1);
                resource.setResourceDesc(param.getDesc());
                resource.setUpdateTime(new Date());
                resource.setUserName(userDomain.getUserName());
                resource.setFileName(param.getOriginalFilename());
                resource.setType(0);
                resource.setFilePath(param.getFilePath());
                resource.setFileDownloadUrl(param.getFileDownloadUrl());
                resource.setNodeId(param.getNodeId());
                resource.setProjectId(projectId);
                resource.setUploadTime(new Date());
                datadevResourceService.save(resource);
            }
            nodeEntity.setName(param.getNodeName());
            final boolean hasSameNode =
                datadevMenuNodeService.hasSameNode(
                    param.getNodeId(), param.getPid(), param.getNodeName(), projectId);
            if (hasSameNode) {
                throw new BusinessException("同名节点已存在");
            }
            nodeEntity.setPid(param.getPid());
            nodeEntity.setSort(maxSort + 1);
            datadevMenuNodeService.updateById(nodeEntity);
            datadevResourceService.updateBatchById(resourceConfig);
            return true;
        }

        final boolean hasSameNode =
            datadevMenuNodeService.hasSameNode(null, param.getPid(), param.getNodeName(), projectId);
        if (hasSameNode) {
            throw new BusinessException("同名节点已存在");
        }

        DatadevMenuNodeEntity offlineDevMenuNodeEntity = new DatadevMenuNodeEntity();
        offlineDevMenuNodeEntity.setNodeType(OfflineDevNodeTypeEnum.RESOURCE.getNodeType());
        offlineDevMenuNodeEntity.setName(param.getNodeName());
        offlineDevMenuNodeEntity.setProjectId(projectId);
        offlineDevMenuNodeEntity.setPid(param.getPid());
        offlineDevMenuNodeEntity.setSort(maxSort + 1);
        offlineDevMenuNodeEntity.setRowState(1);
        offlineDevMenuNodeEntity.setCreateTime(new Date());
        offlineDevMenuNodeEntity.setCreatorId(userId);
        datadevMenuNodeService.save(offlineDevMenuNodeEntity);

        DatadevResourceEntity resource = new DatadevResourceEntity();
        resource.setIsActive(1);
        resource.setResourceDesc(param.getDesc());
        resource.setUpdateTime(new Date());
        resource.setUploadTime(new Date());
        resource.setFileName(param.getOriginalFilename());
        resource.setType(0);
        resource.setNodeId(offlineDevMenuNodeEntity.getId());
        resource.setFilePath(param.getFilePath());
        resource.setUserName(userDomain.getUserName());
        resource.setFileDownloadUrl(param.getFileDownloadUrl());
        resource.setResourceDesc(param.getDesc());
        resource.setUpdateTime(new Date());
        resource.setProjectId(projectId);
        datadevResourceService.save(resource);
        return true;
    }

    @Override
    public OfflineDevNodeDetailDomain getNodeDetail(Long id, Long projectId) {
        final DatadevMenuNodeEntity node = datadevMenuNodeService.getById(id);
        if (null == node) {
            throw new BusinessException("节点不存在");
        }

        final DatadevMenuNodeEntity parentNode = datadevMenuNodeService.getById(node.getPid());
        OfflineDevNodeDetailDomain domain = new OfflineDevNodeDetailDomain();
        if (null == parentNode && TOP_NODE_ID.contains(node.getPid())) {
            domain.setPname(OfflineDevTopNodeIdEnum.of(String.valueOf(node.getPid())).getName());
        } else {
            domain.setPname(parentNode.getName());
        }
        domain.setId(id);
        domain.setPid(node.getPid());
        domain.setNodeType(node.getNodeType());
        domain.setNodeName(node.getName());
        domain.setRemark(node.getRemark());

        return domain;
    }

    private String uploadHdfsFile(String filePath, Long projectId) {

        String username = "hdfs";
        FileSystem hdfs = null;
        String hdfsUrl = hdfsConfig.getHdfsUrl();
        String hdfsNameservices = hdfsConfig.getHdfsNameservices();
        String hdfsNamenodes = hdfsConfig.getHdfsNamenodes();
        String hdfsNamenodesaddr = hdfsConfig.getHdfsNamenodesaddr();
        String hdfsPathDir = null;
        String fileName = null;
        try {
            hdfs =
                HdfsUtil.getFileSystem(
                    null, username, null, hdfsUrl, hdfsNameservices, hdfsNamenodesaddr, hdfsNamenodes);

            hdfsPathDir =
                String.format(
                    "%s%s%s%s%s",
                    File.separator, "jar", File.separator, projectId, File.separator);
            File file = new File(filePath);
            fileName = file.getName();
            Path localPath = new Path(new URI(filePath));
            Path hdfsPath = new Path(hdfsPathDir);
            if (hdfs != null) {
                if (!hdfs.exists(hdfsPath)) {
                    hdfs.mkdirs(hdfsPath);
                }
                hdfs.copyFromLocalFile(localPath, hdfsPath);

                hdfs.close();
            }
            boolean delete = file.delete();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new BusinessException("上传失败,请检查hdfs url");
        }
        String r1 = hdfsPathDir + fileName;
        return r1;
    }

    @Override
    public OfflineDevTaskFlowVNodeDetailDomain getNodeDetailByTaskFlowId(Long id, Long projectId) {
        List<DatadevMenuNodeEntity> multi =
            datadevMenuNodeService.list(
                new LambdaQueryWrapper<DatadevMenuNodeEntity>().eq(DatadevMenuNodeEntity::getPid, id));
        if (CollectionUtils.isEmpty(multi)) {
            return null;
        }
        DatadevMenuNodeEntity vNode =
            multi.stream()
                .filter(e -> e.getNodeType().equals(OfflineDevNodeTypeEnum.VNODE_START.getNodeType()))
                .findFirst()
                .get();
        OfflineDevTaskFlowVNodeDetailDomain domain = new OfflineDevTaskFlowVNodeDetailDomain();
        DatadevVnodeEntity vnodeStartConfig =
            datadevVnodeSerivce.getDatadevVnodeByNodeId(vNode.getId());
        if (null != vnodeStartConfig) {
            domain =
                ConvertUtil.copyProperties(vnodeStartConfig, OfflineDevTaskFlowVNodeDetailDomain.class);
        }
        domain.setName(vNode.getName());
        domain.setNodeType(vNode.getNodeType());
        domain.setNodeTypeName(OfflineDevNodeTypeEnum.of(vNode.getNodeType()).getNodeDesc());
        domain.setCreateTime(vNode.getCreateTime());
        domain.setUpdateTime(vNode.getUpdateTime());
        domain.setRemark(vNode.getRemark());
        return domain;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean lockTask(Long id, Long projectId, String userId, Integer isLock) {
        final DatadevMenuNodeEntity node = datadevMenuNodeService.getById(id);
        if (node == null) {
            throw new BusinessException("节点不存在");
        }

        node.setIsLocked(isLock);
        node.setUpdaterId(userId);

        datadevMenuNodeService.updateById(node);

        String msg;
        if (isLock.equals(1)) {
            msg = String.format("节点:%s被锁定", node.getName());
        } else {
            msg = String.format("节点:%s被解锁", node.getName());
        }

        datadevOperationHistoryService.insertHistory(id, projectId, msg);

        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean saveTaskScript(
        Long id, Long projectId, String userId, String script, List<OfflineDevTaskScriptParam> params) {
        final DatadevMenuNodeEntity node = datadevMenuNodeService.getById(id);
        if (null == node) {
            throw new BusinessException("节点不存在");
        }

        DatadevScriptEntity taskConfig = datadevScriptService.getDatadevScriptByNodeId(id);

        List<Param> paramList = new ArrayList<>();

        if (CollectionUtils.isNotEmpty(params)) {
            paramList =
                params.stream()
                    .map(p -> new Param(p.getParamName(), p.getParamValue()))
                    .collect(Collectors.toList());
        }

        // 解析script，取出target table
        // 现在使用tableproduct表了
    /* DataSourceDomain systemDataSourceByProjectId =
            dataSourceService.getSystemDataSourceByProjectId(projectId);
        String dsType = systemDataSourceByProjectId.getDsType();
        List<String> scripts = Lists.newArrayList();
        if (dsType.equals(DSType.Sap.name())) {
          scripts = SqlParserUtil.sapValidSplitSql(script);
        }else {
          DbType dbType = DbType.of(dsType);
          scripts = SqlParserUtil.split(script,dbType);
        }
        List<String> targetTables = new ArrayList<>();
        for (String sql : scripts) {
          // 遍历scripts中的每一个sql，找到targetTable

          if (Database.SAP.equals(database)) {
            lineageService.analySapLineageAndInsert(sql, projectId, jobEntity);
          } else {
            lineageService.analyseAndInsertSqlLineage(sql, database, projectId, jobEntity);
          }
          List<LineageDomain> lineageDomains = lineageService.analyseSqlLineage(sql, from, projectId);
          if (CollectionUtils.isNotEmpty(lineageDomains)) {
            // a b
            // a c  => a [b,c,d]
            // a d
            // 返回的list只会存在一个targetTable，取第一个就可以了
            targetTables.add(lineageDomains.get(0).getTargetTable());
          }
        }
    */
        if (null == taskConfig) {
            taskConfig = new DatadevScriptEntity();
            taskConfig.setNodeId(id);
            taskConfig.setCreatorId(userId);
            taskConfig.setUpdaterId(userId);

            ScriptTask scriptTask = new ScriptTask();
            scriptTask.setScript(script);
            scriptTask.setParamList(paramList);
            taskConfig.setScript(script);
            taskConfig.setParam(JSONArray.toJSONString(paramList));
            // scriptTask.setTargetTables(targetTables);

            datadevScriptService.save(taskConfig);
            //        LogContext.put(LogKey.NAME.getKey(), node.getName());
        } else {

            String paramsString = JSONArray.toJSONString(paramList);
            taskConfig.setScript(script);
            taskConfig.setParam(paramsString);

            taskConfig.setUpdaterId(userId);
            datadevScriptService.updateById(taskConfig);
            //        LogContext.put(LogKey.ISUPDATE.getKey(), LogKey.ISUPDATE.getValue());
            //        LogContext.put(LogKey.NAME.getKey(), node.getName());
        }

        // todo 解析script，取出source/target table

        // datadevOperationHistoryService.insertHistory(id, projectId, "保存脚本");
        datadevOperationHistoryService.insertHistory2(id, userId, projectId, "保存脚本");

        return true;
    }

    @Override
    public OfflineDevTaskScriptDetailDomain getTaskScript(Long id, Long projectId) {
        final DatadevMenuNodeEntity node = datadevMenuNodeService.getById(id);
        if (null == node) {
            throw new BusinessException("节点不存在");
        }

        OfflineDevTaskScriptDetailDomain domain = new OfflineDevTaskScriptDetailDomain();
        domain.setIsLocked(node.getIsLocked());

        DatadevScriptEntity scriptTaskConfig = datadevScriptService.getDatadevScriptByNodeId(id);
        if (null == scriptTaskConfig) {
            return domain;
        }

        domain.setScript(scriptTaskConfig.getScript());

        String param = scriptTaskConfig.getParam();
        List<Param> paramList = JSONArray.parseArray(param, Param.class);
        if (CollectionUtils.isNotEmpty(paramList)) {

            final List<OfflineDevTaskScriptParam> params =
                paramList.stream()
                    .map(p -> new OfflineDevTaskScriptParam(p.getParamName(), p.getParamValue()))
                    .collect(Collectors.toList());
            domain.setParams(params);
        }

        return domain;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean cloneTask(
        Long id, Long taskFlowNodeId, String newNodeName, Long projectId, String userId) {
        final DatadevMenuNodeEntity targetNode = datadevMenuNodeService.getById(id);

        if (null == targetNode) {
            throw new BusinessException("节点不存在");
        }

        Integer nodeType = targetNode.getNodeType();
        // todo 通过类型去获取相关配置进行复制

        final OfflineDevNodeTypeEnum targetNodeType =
            OfflineDevNodeTypeEnum.of(targetNode.getNodeType());

        // 克隆到某个工作流
        if (Objects.nonNull(taskFlowNodeId)) {
            final DatadevMenuNodeEntity taskFlowNode = datadevMenuNodeService.getById(taskFlowNodeId);

            if (null == taskFlowNode) {
                throw new BusinessException("节点不存在");
            }

            Integer taskFlowDefaultNodeType;

            if (targetNodeType.isDataintegrationTask()) {
                taskFlowDefaultNodeType = OfflineDevNodeTypeEnum.DIR_DATAINTEGRATION.getNodeType();
            } else if (targetNodeType.isDatadevTask()) {
                taskFlowDefaultNodeType = OfflineDevNodeTypeEnum.DIR_DATADEV.getNodeType();
            } else {
                taskFlowDefaultNodeType = OfflineDevNodeTypeEnum.DIR_DATAQUALITY.getNodeType();
            }

            DatadevMenuNodeEntity taskFlowDefaultNode =
                getByTaskFlowNodeIdAndDefaultNodeType(taskFlowNodeId, taskFlowDefaultNodeType);

            final boolean hasSameNode =
                datadevMenuNodeService.hasSameNode(
                    null, taskFlowDefaultNode.getId(), newNodeName, projectId);
            if (hasSameNode) {
                throw new BusinessException("已存在相同名称的节点");
            }

            final Integer maxSort = datadevMenuNodeService.getMaxSortByPid(taskFlowDefaultNode.getId());

            // 克隆节点
            final DatadevMenuNodeEntity newNode =
                ConvertUtil.copyProperties(targetNode, DatadevMenuNodeEntity.class);
            newNode.setName(newNodeName);
            newNode.setPid(taskFlowDefaultNode.getId());
            newNode.setSort(maxSort + 1);
            newNode.setCreatorId(userId);
            newNode.setUpdaterId(userId);
            datadevMenuNodeService.save(newNode);

            // 克隆配置
            //      if (null != targetNodeConfig) {
            ////        targetNodeConfig.setNodeId(newNode.getId());
            ////        targetNodeConfig.setCreatorId(userId);
            ////        targetNodeConfig.setUpdaterId(userId);
            ////        targetNodeConfig.setId(null);
            ////        offlineDevTaskConfigService.insertOne(targetNodeConfig);
            //      }

            // 将克隆的节点放置到工作流配置中

            List<DatadevSubNodeEntity> subTaskList =
                datadevSubNodeService.getDatadevSubNodeList(taskFlowNodeId);
            final String maxCoordX =
                subTaskList.stream()
                    .max(
                        (st1, st2) ->
                            new BigDecimal(st1.getCoordX()).compareTo(new BigDecimal(st2.getCoordX())))
                    .get()
                    .getCoordX();
            final String maxCoordY =
                subTaskList.stream()
                    .max(
                        (st1, st2) ->
                            new BigDecimal(st1.getCoordY()).compareTo(new BigDecimal(st2.getCoordY())))
                    .get()
                    .getCoordY();

            DatadevSubNodeEntity subTask = new DatadevSubNodeEntity();
            subTask.setNodeId(newNode.getId());
            subTask.setRowState(1);
            subTask.setCoordX(new BigDecimal(maxCoordX).add(new BigDecimal("50")).toString());
            subTask.setCoordY(new BigDecimal(maxCoordY).add(new BigDecimal("50")).toString());
            subTaskList.add(subTask);
            subTaskList.forEach(
                e -> {
                    e.setProjectId(projectId);
                    e.setCreatorId(userId);
                    e.setUpdateTime(e.getUpdateTime());
                });

            datadevSubNodeService.saveOrUpdateBatch(subTaskList);

            datadevOperationHistoryService.insertHistory(
                newNode.getId(), projectId, String.format("克隆任务:%s", newNodeName));

            return true;
        }

        // 克隆到非工作流（作业开发）
        final Integer maxSort = datadevMenuNodeService.getMaxSortByPid(targetNode.getPid());

        final boolean hasSameNode =
            datadevMenuNodeService.hasSameNode(null, targetNode.getPid(), newNodeName, projectId);
        if (hasSameNode) {
            throw new BusinessException("已存在相同名称的节点");
        }

        // 克隆节点
        final DatadevMenuNodeEntity newNode =
            ConvertUtil.copyProperties(targetNode, DatadevMenuNodeEntity.class);
        newNode.setName(newNodeName);
        newNode.setPid(targetNode.getPid());
        newNode.setSort(maxSort + 1);
        newNode.setCreatorId(userId);
        newNode.setUpdaterId(userId);
        newNode.setId(null);
        datadevMenuNodeService.save(newNode);

        // 克隆配置
        //    if (null != targetNodeConfig) {
        //      targetNodeConfig.setNodeId(newNode.getId());
        //      targetNodeConfig.setCreatorId(userId);
        //      targetNodeConfig.setUpdaterId(userId);
        //      targetNodeConfig.setId(null);
        //      offlineDevTaskConfigService.insertOne(targetNodeConfig);
        //    }

        datadevOperationHistoryService.insertHistory(
            newNode.getId(), projectId, String.format("克隆任务:%s", newNodeName));

        return true;
    }

    @Override
    public PageResult<OfflineDevOperationHistoryDomain> getTaskOperationHistory(
        Long id, String content, Integer pageNum, Integer pageSize, Long projectId) {
        final DatadevMenuNodeEntity node = datadevMenuNodeService.getById(id);
        if (null == node) {
            throw new BusinessException("节点不存在");
        }

        List<Long> nodeIds = Lists.newArrayList(id);

        final OfflineDevNodeTypeEnum nodeTypeEnum = OfflineDevNodeTypeEnum.of(node.getNodeType());

        // 如果是工作流，需要查询工作流下面所有工作组件节点的历史记录
        if (nodeTypeEnum.equals(OfflineDevNodeTypeEnum.DIR_TASK_FLOW)) {
            List<DatadevSubNodeEntity> datadevSubNodeList =
                datadevSubNodeService.getDatadevSubNodeList(id);
            if (CollectionUtils.isNotEmpty(datadevSubNodeList)) {
                datadevSubNodeList.forEach(st -> nodeIds.add(st.getNodeId()));
            }
        }

        final IPage<OfflineDevOperationHistoryDomain> page =
            datadevOperationHistoryService
                .pageQuery(nodeIds, content, pageNum, pageSize)
                .convert(e -> Convert.convert(OfflineDevOperationHistoryDomain.class, e));

        return PageResult.convert(page);
    }

    @Override
    public OfflineDevTaskTableInfoDomain taskTableList(Long projectId) {

        String key = String.format(RedisKey.DATADEV_SQL_KEYWORD, projectId);
        String value = redisTemplate.opsForValue().get(key);
        if (StringUtils.isBlank(value)) {
            return new OfflineDevTaskTableInfoDomain();
        }
        OfflineDevTaskTableInfoDomain offlineDevTaskTableInfoDomain =
            JSONObject.parseObject(value, OfflineDevTaskTableInfoDomain.class);
        return offlineDevTaskTableInfoDomain;
    }

    //    @BizDataIndexAnnotation(
//        type = BizDataIndexTypeEnum.OFFLINE_TASKFLOW,
//        dataIdFieldName = "id",
//        dataIdParamIndex = 0,
//        operateType = BizDataIndexAnnotation.OperateType.SAVE)
    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public DatadevMenuNodeEntity saveTaskFlow(OfflineDevSaveTaskFlowParam param) {
        final Long id = param.getId();
        final Long pid = param.getPid();
        final String name = param.getName();
        final String remark = param.getRemark();
        final Long projectId = param.getProjectId();
        final String userId = param.getUserId();

        DatadevMenuNodeEntity entity = new DatadevMenuNodeEntity();
        entity.setId(id);
        entity.setPid(pid);
        entity.setName(name);
        entity.setRemark(remark);
        entity.setUpdaterId(userId);

        Integer maxSort = datadevMenuNodeService.getMaxSortByPid(pid);
        maxSort = (maxSort == null) ? 0 : maxSort;
        // 修改
        if (null != id) {
            final DatadevMenuNodeEntity nodeEntity = datadevMenuNodeService.getById(id);
            if (null == nodeEntity) {
                throw new BusinessException("节点不存在");
            }

            final boolean hasSameNode = datadevMenuNodeService.hasSameNode(id, pid, name, projectId);
            if (hasSameNode) {
                throw new BusinessException("同名节点已存在");
            }

            // 如果任务名称不一样，需要批量修改运维中心的任务名称
            if (!name.equals(nodeEntity.getName())) {
                LambdaUpdateWrapper<DatadevJobEntity> update =
                    new UpdateWrapper<DatadevJobEntity>().lambda();
                update.eq(DatadevJobEntity::getNodeId, id);
                update.set(DatadevJobEntity::getNodeName, name);
                datadevJobService.update(update);
            }

            if (!nodeEntity.getPid().equals(pid)) {
                nodeEntity.setSort(maxSort + 1);
            }
            nodeEntity.setName(name);
            nodeEntity.setRemark(remark);
            nodeEntity.setUpdateTime(new Date());
            nodeEntity.setPid(pid);
            datadevMenuNodeService.updateById(nodeEntity);
            //        LogContext.put(LogKey.ISUPDATE.getKey(), LogKey.ISUPDATE.getValue());
            //        LogContext.put(LogKey.NAME.getKey(), entity.getName());

            datadevOperationHistoryService.insertHistory(
                id, projectId, String.format("修改工作流节点:[%s]的配置", name));

            return nodeEntity;
        }

        // 新增
        final boolean hasSameNode = datadevMenuNodeService.hasSameNode(null, pid, name, projectId);
        if (hasSameNode) {
            throw new BusinessException("同名节点已存在");
        }
        entity.setCreatorId(userId);
        entity.setProjectId(projectId);
        entity.setNodeType(OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType());
        entity.setRowState(1);
        entity.setSort(maxSort + 1);
        entity.setIsLocked(0);

        datadevMenuNodeService.save(entity);

        param.setId(entity.getId());

        // 创建开始节点
        DatadevMenuNodeEntity vnode = new DatadevMenuNodeEntity();
        vnode.setName("开始节点");
        vnode.setPid(entity.getId());
        vnode.setProjectId(projectId);
        vnode.setSort(1);
        vnode.setNodeType(OfflineDevNodeTypeEnum.VNODE_START.getNodeType());
        vnode.setRowState(1);
        vnode.setIsSystem(1);
        vnode.setCreatorId(userId);
        vnode.setUpdaterId(userId);
        datadevMenuNodeService.save(vnode);

        // 将开始节点加入到工作流的子节点
        //    OfflineDevTaskConfigEntity configEntity = new OfflineDevTaskConfigEntity();
        //    configEntity.setProjectId(projectId);
        //    configEntity.setNodeId(entity.getId());
        //    configEntity.setVnodeStartConfig(new OfflineDevTaskConfigEntity.VnodeStart());
        //    configEntity.setCreatorId(userId);
        //    configEntity.setUpdaterId(userId);
        DatadevVnodeEntity datadevVnodeEntity = new DatadevVnodeEntity();
        datadevVnodeEntity.setNodeId(vnode.getId());
        datadevVnodeEntity.setCreateTime(new Date());
        datadevVnodeEntity.setCreatorId(userId);
        datadevVnodeEntity.setUpdaterId(userId);
        datadevVnodeEntity.setUpdateTime(new Date());
        datadevVnodeSerivce.save(datadevVnodeEntity);

        DatadevSubNodeEntity taskFlowSubTask = new DatadevSubNodeEntity();
        taskFlowSubTask.setNodeId(vnode.getId());
        taskFlowSubTask.setRowState(1);
        taskFlowSubTask.setCoordX("521");
        taskFlowSubTask.setCoordY("145");
        taskFlowSubTask.setTaskFlowId(entity.getId());

        datadevSubNodeService.save(taskFlowSubTask);

        // 生成固定的3个子目录
        List<DatadevMenuNodeEntity> defaultEntities = new ArrayList<>();

        DatadevMenuNodeEntity integrationEntity = new DatadevMenuNodeEntity();
        integrationEntity.setName("数据集成");
        integrationEntity.setPid(entity.getId());
        integrationEntity.setProjectId(projectId);
        integrationEntity.setSort(2);
        integrationEntity.setNodeType(OfflineDevNodeTypeEnum.DIR_DATAINTEGRATION.getNodeType());
        integrationEntity.setRowState(1);
        integrationEntity.setIsSystem(1);
        integrationEntity.setCreatorId(userId);
        integrationEntity.setUpdaterId(userId);

        DatadevMenuNodeEntity datadevEntity = new DatadevMenuNodeEntity();
        datadevEntity.setName("数据开发");
        datadevEntity.setPid(entity.getId());
        datadevEntity.setProjectId(projectId);
        datadevEntity.setSort(3);
        datadevEntity.setNodeType(OfflineDevNodeTypeEnum.DIR_DATADEV.getNodeType());
        datadevEntity.setRowState(1);
        datadevEntity.setIsSystem(1);
        datadevEntity.setCreatorId(userId);
        datadevEntity.setUpdaterId(userId);

        DatadevMenuNodeEntity dataqualityEntity = new DatadevMenuNodeEntity();
        dataqualityEntity.setName("数据质量");
        dataqualityEntity.setPid(entity.getId());
        dataqualityEntity.setProjectId(projectId);
        dataqualityEntity.setSort(4);
        dataqualityEntity.setNodeType(OfflineDevNodeTypeEnum.DIR_DATAQUALITY.getNodeType());
        dataqualityEntity.setRowState(1);
        dataqualityEntity.setIsSystem(1);
        dataqualityEntity.setCreatorId(userId);
        dataqualityEntity.setUpdaterId(userId);

        defaultEntities.add(integrationEntity);
        defaultEntities.add(datadevEntity);
        defaultEntities.add(dataqualityEntity);

        datadevMenuNodeService.saveBatch(defaultEntities);
//        LogContext.put(LogKey.NAME.getKey(), entity.getName());
        return entity;
    }

    @Override
    public OfflineDevTaskFlowDetailDomain getTaskFlowDetail(Long id, Long projectId) {
        OfflineDevTaskFlowDetailDomain domain = new OfflineDevTaskFlowDetailDomain();

        final DatadevMenuNodeEntity menuNode = datadevMenuNodeService.getById(id);
        if (null == menuNode) {
            throw new BusinessException("节点不存在");
        }

        List<OfflineDevTaskFlowDetailDomain.Node> nodes = new ArrayList<>();
        List<OfflineDevTaskFlowDetailDomain.Edge> edges = new ArrayList<>();

        domain.setNodeId(id);
        domain.setIsLock(menuNode.getIsLocked());
        domain.setXxlJobId(menuNode.getXxlJobId());

        List<DatadevSubNodeEntity> r1 = datadevSubNodeService.getDatadevSubNodeList(id);

        List<DatadevSubNodeDependencyEntity> subTaskDependencyList =
            datadevSubNodeDependencyService.getDatadevSubNodeDepenDencyList(id);
        final List<DatadevSubNodeEntity> subTaskList =
            r1.stream().filter(st -> st.getRowState() == 1).collect(Collectors.toList());

        final List<Long> subTaskNodeIds =
            subTaskList.stream().map(st -> st.getNodeId()).collect(Collectors.toList());

        List<DatadevMenuNodeEntity> subTaskNodes = datadevMenuNodeService.listByIds(subTaskNodeIds);

        final Map<Long, DatadevMenuNodeEntity> subTaskNodeMap =
            subTaskNodes.stream().collect(Collectors.toMap(x -> x.getId(), x -> x));

        for (DatadevSubNodeEntity taskFlowSubTask : subTaskList) {
            if (taskFlowSubTask.getRowState().equals(0)) {
                continue;
            }
            final DatadevMenuNodeEntity entity = subTaskNodeMap.get(taskFlowSubTask.getNodeId());

            final OfflineDevNodeTypeEnum nodeTypeEnum = OfflineDevNodeTypeEnum.of(entity.getNodeType());

            final OfflineDevTaskFlowDetailDomain.Node node =
                OfflineDevTaskFlowDetailDomain.Node.builder()
                    .id(taskFlowSubTask.getNodeId())
                    .nodeName(entity.getName())
                    .nodeType(entity.getNodeType())
                    .apiType(entity.getApiType())
                    .x(taskFlowSubTask.getCoordX())
                    .y(taskFlowSubTask.getCoordY())
                    .supportClone(!OfflineDevNodeTypeEnum.VNODE_START.equals(nodeTypeEnum))
                    .build();

            nodes.add(node);
        }

        DatadevMenuNodeEntity vnodeByTaskFlowId =
            datadevMenuNodeService.getDatadevMenuNodeByFlowId(id, OfflineDevNodeTypeEnum.VNODE_START);

        if (CollectionUtils.isNotEmpty(subTaskDependencyList)) {
            for (DatadevSubNodeDependencyEntity line : subTaskDependencyList) {

                OfflineDevTaskFlowDetailDomain.Edge edge = new OfflineDevTaskFlowDetailDomain.Edge();
                edge.setFrom(line.getUpNodeId());
                edge.setTo(line.getDownNodeId());
                edge.setStatus(line.getStatus());
                edge.setFixed(1);
                if (line.getUpNodeId().equals(vnodeByTaskFlowId.getId())) {
                    edge.setFixed(0);
                }
                edges.add(edge);
            }
        }

        domain.setNodes(nodes);
        domain.setEdges(edges);

        return domain;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public OfflineDevTaskFlowDetailDomain.Node saveTaskFlowNode(
        OfflineDevTaskFlowNodeSaveParam request, Long projectId, String userId) {
        final DatadevMenuNodeEntity taskFlowNode = datadevMenuNodeService.getById(request.getId());
        if (null == taskFlowNode) {
            throw new BusinessException("节点不存在");
        }

        // 如果仅仅是更改位置
        if (Objects.nonNull(request.getSubTaskId())) {
            List<DatadevSubNodeEntity> subTaskList =
                datadevSubNodeService.getDatadevSubNodeList(request.getId());
            for (DatadevSubNodeEntity taskFlowSubTask : subTaskList) {
                if (Objects.isNull(taskFlowSubTask.getNodeId())) {
                    continue;
                }
                if (taskFlowSubTask.getNodeId().equals(request.getSubTaskId())) {
                    taskFlowSubTask.setCoordX(request.getX());
                    taskFlowSubTask.setCoordY(request.getY());
                    break;
                }
            }

            DatadevMenuNodeEntity offlineDevMenuNode =
                datadevMenuNodeService.getById(request.getSubTaskId());

            if (!offlineDevMenuNode.getName().equals(request.getNodeName())
                || !Objects.equals(request.getRemark(), offlineDevMenuNode.getRemark())) {
                OfflineDevNodeTypeEnum of = OfflineDevNodeTypeEnum.of(request.getNodeType());
                String msg = String.format("修改%s节点:%s", of.getNodeDesc(), request.getNodeName());
                datadevOperationHistoryService.insertHistory(request.getSubTaskId(), projectId, msg);
            }

            offlineDevMenuNode.setName(request.getNodeName());
            offlineDevMenuNode.setRemark(request.getRemark());
            datadevMenuNodeService.updateById(offlineDevMenuNode);

            datadevSubNodeService.updateBatchById(subTaskList);
            OfflineDevTaskFlowDetailDomain.Node node =
                OfflineDevTaskFlowDetailDomain.Node.builder()
                    .id(request.getSubTaskId())
                    .nodeName(request.getNodeName())
                    .nodeType(request.getNodeType())
                    .apiType(request.getApiType())
                    .x(request.getX())
                    .y(request.getY())
                    .supportClone(true)
                    .build();
            //        LogContext.put(LogKey.NAME.getKey(), request.getNodeName());
            //        LogContext.put(LogKey.ISUPDATE.getKey(), LogKey.ISUPDATE.getValue());
            return node;
        }

        Integer defaultNodeType;

        final OfflineDevNodeTypeEnum nodeTypeEnum = OfflineDevNodeTypeEnum.of(request.getNodeType());
        if (nodeTypeEnum.isDataintegrationTask()) {
            defaultNodeType = OfflineDevNodeTypeEnum.DIR_DATAINTEGRATION.getNodeType();
        } else if (nodeTypeEnum.isDatadevTask()) {
            defaultNodeType = OfflineDevNodeTypeEnum.DIR_DATADEV.getNodeType();
        } else {
            defaultNodeType = OfflineDevNodeTypeEnum.DIR_DATAQUALITY.getNodeType();
        }

        final DatadevMenuNodeEntity defaultNode =
            getByTaskFlowNodeIdAndDefaultNodeType(request.getId(), defaultNodeType);

        // 创建节点
        DatadevMenuNodeEntity entity = new DatadevMenuNodeEntity();
        entity.setPid(defaultNode.getId());
        entity.setName(request.getNodeName());
        entity.setRemark(request.getRemark());
        entity.setUpdaterId(userId);
        entity.setCreatorId(userId);
        entity.setProjectId(projectId);
        entity.setNodeType(request.getNodeType());
        entity.setApiType(request.getApiType());
        entity.setRowState(1);

        final boolean hasSameNode =
            datadevMenuNodeService.hasSameNode(
                null, defaultNode.getId(), request.getNodeName(), projectId);
        if (hasSameNode) {
            throw new BusinessException("同名节点已存在");
        }

        final Integer maxSort = datadevMenuNodeService.getMaxSortByPid(defaultNode.getId());

        entity.setSort(maxSort + 1);

        datadevMenuNodeService.save(entity);

        // 放置到工作流子节点中
        //    final OfflineDevTaskConfigEntity taskConfig =
        //        offlineDevTaskConfigService.getByNodeId(request.getId());
        //
        //    final List<OfflineDevTaskConfigEntity.TaskFlowSubTask> subTaskList =
        //        taskConfig.getTaskFlowConfig().getSubTaskList();

        DatadevSubNodeEntity subTask = new DatadevSubNodeEntity();
        subTask.setNodeId(entity.getId());
        subTask.setRowState(1);
        subTask.setCoordX(request.getX());
        subTask.setCoordY(request.getY());
        subTask.setTaskFlowId(request.getId());
        datadevSubNodeService.save(subTask);
//        LogContext.put(LogKey.NAME.getKey(), entity.getName());

        String msg = String.format("新增%s节点:%s", nodeTypeEnum.getNodeDesc(), request.getNodeName());

        datadevOperationHistoryService.insertHistory(entity.getId(), projectId, msg);

        final OfflineDevTaskFlowDetailDomain.Node node =
            OfflineDevTaskFlowDetailDomain.Node.builder()
                .id(entity.getId())
                .nodeName(request.getNodeName())
                .nodeType(request.getNodeType())
                .apiType(request.getApiType())
                .x(request.getX())
                .y(request.getY())
                .supportClone(true)
                .build();

        return node;
    }

    @Override
    public Boolean saveTaskFlowNodeRelation(
        Long taskFlowNodeId,
        Long upNodeId,
        Long downNodeId,
        Integer status,
        Long projectId,
        String userId) {

        final DatadevMenuNodeEntity taskFlowNode = datadevMenuNodeService.getById(taskFlowNodeId);
        if (null == taskFlowNode) {
            throw new BusinessException("节点不存在");
        }

        final DatadevMenuNodeEntity upNode = datadevMenuNodeService.getById(upNodeId);

        List<DatadevSubNodeDependencyEntity> subTaskDependencyList =
            datadevSubNodeDependencyService.getDatadevSubNodeDepenDencyList(taskFlowNodeId);

        if (CollectionUtils.isNotEmpty(subTaskDependencyList)) {
            if (this.checkCycleDependency(subTaskDependencyList, upNodeId, downNodeId)) {
                throw new BusinessException("不允许有循环依赖");
            }
        }

        final long count1 =
            subTaskDependencyList.stream().filter(d -> d.getDownNodeId().equals(upNodeId)).count();
        if (count1 == 0) {
            if (!Objects.equals(upNode.getNodeType(), OfflineDevNodeTypeEnum.VNODE_START.getNodeType())) {
                throw new BusinessException(
                    String.format("任务[%s]须先有上级连线", upNode.getName()));
            }
        }

        // 如果上游是数据质量，则强制给连线类型为0
        if (upNode.getNodeType().equals(OfflineDevNodeTypeEnum.DATAQUALITY.getNodeType())) {
            status = 0;
        }

        boolean isInsert = true;

        final Iterator<DatadevSubNodeDependencyEntity> iterator = subTaskDependencyList.iterator();
        while (iterator.hasNext()) {
            final DatadevSubNodeDependencyEntity next = iterator.next();
            if (next.getUpNodeId().equals(upNodeId) && next.getDownNodeId().equals(downNodeId)) {
                iterator.remove();
                isInsert = false;
            }
        }
        // 判断是否存在老的连线

        LambdaQueryWrapper<DatadevSubNodeDependencyEntity> eq =
            new LambdaQueryWrapper<DatadevSubNodeDependencyEntity>()
                .eq(DatadevSubNodeDependencyEntity::getTaskFlowId, taskFlowNodeId)
                .eq(DatadevSubNodeDependencyEntity::getUpNodeId, upNodeId)
                .eq(DatadevSubNodeDependencyEntity::getDownNodeId, downNodeId);
        DatadevSubNodeDependencyEntity dependency = datadevSubNodeDependencyService.getOne(eq, false);

        if (Objects.isNull(dependency)) {
            dependency = new DatadevSubNodeDependencyEntity();
        }
        dependency.setUpNodeId(upNodeId);
        dependency.setDownNodeId(downNodeId);
        dependency.setStatus(status);
        dependency.setRowState(1);
        dependency.setCreateTime(new Date());
        dependency.setCreatorId(userId);
        dependency.setUpdateTime(new Date());
        dependency.setTaskFlowId(taskFlowNodeId);
        dependency.setUpdaterId(userId);
        datadevSubNodeDependencyService.saveOrUpdate(dependency);
        if (isInsert) {
            datadevOperationHistoryService.insertHistory(taskFlowNodeId, projectId, "新增节点连线");
        } else {
            datadevOperationHistoryService.insertHistory(taskFlowNodeId, projectId, "修改节点连线");
        }
//        LogContext.put(LogKey.NAME.getKey(), taskFlowNode.getName());
        return true;
    }

    //    @BizDataIndexAnnotation(
//        type = BizDataIndexTypeEnum.OFFLINE_TASKFLOW,
//        dataIdFieldName = "taskFlowNodeId",
//        dataIdParamIndex = 0,
//        operateType = BizDataIndexAnnotation.OperateType.DELETE)
    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean deleteTaskFlowNodeRelation(
        Long taskFlowNodeId, Long upNodeId, Long downNodeId, Long projectId, String userId) {
        final DatadevMenuNodeEntity taskFlowNode = datadevMenuNodeService.getById(taskFlowNodeId);
        if (null == taskFlowNode) {
            throw new BusinessException("节点不存在");
        }

        List<DatadevSubNodeEntity> subTaskList =
            datadevSubNodeService.getDatadevSubNodeList(taskFlowNodeId);
        List<Long> strings =
            subTaskList.stream()
                .filter(
                    e ->
                        Objects.nonNull(e.getNodeId())
                            && Objects.nonNull(e.getRowState())
                            && e.getRowState().equals(0))
                .map(e -> e.getNodeId())
                .collect(Collectors.toList());

        List<DatadevSubNodeDependencyEntity> subTaskDependencyList =
            datadevSubNodeDependencyService.getDatadevSubNodeDepenDencyList(taskFlowNodeId);

        final Iterator<DatadevSubNodeDependencyEntity> iterator = subTaskDependencyList.iterator();

        // 首先查询下下游节点是否是其他节点的上游节点，如果是不能删除

        for (DatadevSubNodeDependencyEntity taskFlowSubTaskDependency : subTaskDependencyList) {
            Long upNodeId1 = taskFlowSubTaskDependency.getUpNodeId();
            Integer rowState = taskFlowSubTaskDependency.getRowState();
            if (upNodeId1.equals(downNodeId)
                && Objects.nonNull(taskFlowSubTaskDependency.getDownNodeId())
                && !Objects.equals(rowState, 0)
                && !strings.contains(upNodeId1)) {
                DatadevMenuNodeEntity byId = datadevMenuNodeService.getById(upNodeId1);
                String logInfo = String.format("请先删除%s节点的下游节点的连线", byId.getName());
                throw new BusinessException(logInfo);
            }
        }

        while (iterator.hasNext()) {
            final DatadevSubNodeDependencyEntity next = iterator.next();
            if (next.getUpNodeId().equals(upNodeId) && next.getDownNodeId().equals(downNodeId)) {
                datadevSubNodeDependencyService.removeById(next.getId());
                iterator.remove();
            }
        }

        datadevOperationHistoryService.insertHistory(taskFlowNodeId, projectId, "删除节点连线");

        return true;
    }

    @Override
    public OfflineDevTaskFlowVNodeDetailDomain getTaskFlowVNodeDetail(Long nodeId, Long projectId) {
        final DatadevMenuNodeEntity vNode = datadevMenuNodeService.getById(nodeId);
        if (null == vNode) {
            throw new BusinessException("节点不存在");
        }

        final UserDomain createUser = userService.getUserById(vNode.getCreatorId());
        final UserDomain updateUser = userService.getUserById(vNode.getUpdaterId());

        OfflineDevTaskFlowVNodeDetailDomain domain = new OfflineDevTaskFlowVNodeDetailDomain();

        DatadevVnodeEntity vnodeStartConfig = datadevVnodeSerivce.getDatadevVnodeByNodeId(nodeId);

        if (null != vnodeStartConfig) {
            domain =
                ConvertUtil.copyProperties(vnodeStartConfig, OfflineDevTaskFlowVNodeDetailDomain.class);
        }

        domain.setName(vNode.getName());
        domain.setNodeType(vNode.getNodeType());
        domain.setNodeTypeName(OfflineDevNodeTypeEnum.of(vNode.getNodeType()).getNodeDesc());
        domain.setCreator(createUser == null ? null : createUser.getUserName());
        domain.setCreateTime(vNode.getCreateTime());
        domain.setModifier(updateUser == null ? null : updateUser.getUserName());
        domain.setUpdateTime(vNode.getUpdateTime());
        domain.setRemark(vNode.getRemark());

        return domain;
    }

    @Override
    public OfflineDevTaskPropertyDomain getTaskFlowNodeDetail(Long nodeId, Long projectId) {
        final DatadevMenuNodeEntity node = datadevMenuNodeService.getById(nodeId);
        if (null == node) {
            throw new BusinessException("节点不存在");
        }

        OfflineDevTaskPropertyDomain domain = new OfflineDevTaskPropertyDomain();

        final UserDomain createUser = userService.getUserById(node.getCreatorId());
        final UserDomain updateUser = userService.getUserById(node.getUpdaterId());

        domain.setName(node.getName());
        domain.setNodeType(node.getNodeType());
        domain.setApiType(node.getApiType());
        domain.setNodeTypeName(OfflineDevNodeTypeEnum.of(node.getNodeType()).getNodeDesc());
        domain.setCreator(createUser == null ? null : createUser.getUserName());
        domain.setCreateTime(node.getCreateTime());
        domain.setModifier(updateUser == null ? null : updateUser.getUserName());
        domain.setUpdateTime(node.getUpdateTime());
        domain.setRemark(node.getRemark());
        domain.setXxlJobId(node.getXxlJobId());

        return domain;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean saveTaskFlowVNodeConfig(Long nodeId, String remark, Long projectId, String userId) {
        final DatadevMenuNodeEntity node = datadevMenuNodeService.getById(nodeId);
        if (null == node) {
            throw new BusinessException("节点不存在");
        }

        node.setRemark(remark);
        node.setUpdaterId(userId);

        datadevMenuNodeService.updateById(node);

        datadevOperationHistoryService.insertHistory(nodeId, projectId, "修改开始节点描述");

        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean saveTaskFlowSchedule(
        OfflineDevTaskFlowVNodeConfigSaveParam param, Long projectId, String userId) {
        final DatadevMenuNodeEntity node = datadevMenuNodeService.getById(param.getTaskFlowId());

        DatadevMenuNodeEntity vnodeId = datadevMenuNodeService.getById(param.getNodeId());
        if (null == vnodeId) {
            throw new BusinessException("节点不存在");
        }

        DatadevVnodeEntity vnodeStartConfig =
            datadevVnodeSerivce.getDatadevVnodeByNodeId(vnodeId.getId());

        if (null == vnodeStartConfig) {
            vnodeStartConfig = new DatadevVnodeEntity();
        }

        vnodeStartConfig.setEffectiveDateStart(param.getEffectiveDateStart());
        vnodeStartConfig.setEffectiveDateEnd(param.getEffectiveDateEnd());
        vnodeStartConfig.setCycle(param.getCycle());
        vnodeStartConfig.setNodeId(vnodeId.getId());
        vnodeStartConfig.setStartTimeHour(param.getStartTimeHour());
        vnodeStartConfig.setStartTimeMin(param.getStartTimeMin());
        vnodeStartConfig.setEndTimeHour(param.getEndTimeHour());
        vnodeStartConfig.setEndTimeMin(param.getEndTimeMin());
        vnodeStartConfig.setExecPoint(param.getExecPoint());
        vnodeStartConfig.setCycleDependency(param.getCycleDependency());

        /** 1.计算下次计划执行时间 */
        Date lastExecutionTime = null;
        DatadevJobEntity taskFlowLastRecord =
            datadevJobService.findTaskFlowLastRecord(param.getTaskFlowId());
        if (null != taskFlowLastRecord) {
            lastExecutionTime = taskFlowLastRecord.getPlanTime();
        }

        Date nextExecutionTime =
            ScheduleCycleUtil.getNextExecutionTime(
                vnodeStartConfig.getCycle(),
                vnodeStartConfig.getExecPoint(),
                vnodeStartConfig.getEffectiveDateStart(),
                vnodeStartConfig.getEffectiveDateEnd(),
                lastExecutionTime,
                vnodeStartConfig.getStartTimeHour(),
                vnodeStartConfig.getStartTimeMin(),
                vnodeStartConfig.getEndTimeHour(),
                vnodeStartConfig.getEndTimeMin());

        vnodeStartConfig.setNextExecTime(nextExecutionTime);

        datadevVnodeSerivce.saveOrUpdate(vnodeStartConfig);

        final Integer xxlJobId = saveXxlJob(param.getNodeId());
        if (node.getXxlJobId() == null) {
            node.setXxlJobId(xxlJobId);
            node.setUpdaterId(userId);
            node.setPublishTime(new Date());
            datadevMenuNodeService.updateById(node);

            vnodeId.setUpdaterId(userId);
            vnodeId.setUpdateTime(new Date());
            datadevMenuNodeService.updateById(vnodeId);
        }

//        LogContext.put(LogKey.NAME.getKey(), node.getName());
//        LogContext.put(LogKey.ISUPDATE.getKey(), LogKey.ISUPDATE.getValue());
        this.setTaskFlowScheduleStatus(param.getTaskFlowId(), 1);
        datadevOperationHistoryService.insertHistory(param.getNodeId(), projectId, "修改调度配置");

//        bizDataIndexService.saveIndexEs(
//            BizDataIndexTypeEnum.OFFLINE_TASKFLOW, projectId, vnodeId.getPid());

        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean saveTaskFlowNodeConfig(
        Long nodeId, String nodeName, String remark, Long projectId, String userId) {
        final DatadevMenuNodeEntity node = datadevMenuNodeService.getById(nodeId);
        if (null == node) {
            throw new BusinessException("节点不存在");
        }

        node.setName(nodeName);
        node.setRemark(remark);
        node.setUpdaterId(userId);

        datadevMenuNodeService.updateById(node);

        datadevOperationHistoryService.insertHistory(nodeId, projectId, "修改节点配置");

//        LogContext.put(LogKey.NAME.getKey(), node.getName());
//        LogContext.put(LogKey.ISUPDATE.getKey(), LogKey.ISUPDATE.getValue());
        return true;
    }

    @Override
    public List<OfflineDevTaskFlowDomain> getTaskFlowList(Long projectId) {
        LambdaQueryWrapper<DatadevMenuNodeEntity> lambdaQueryWrapper =
            new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                .eq(DatadevMenuNodeEntity::getProjectId, projectId)
                .eq(DatadevMenuNodeEntity::getRowState, 1)
                .in(
                    DatadevMenuNodeEntity::getNodeType,
                    Lists.newArrayList(
                        OfflineDevNodeTypeEnum.DIR_NORMAL.getNodeType(),
                        OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType()));

        lambdaQueryWrapper.orderByDesc(
            DatadevMenuNodeEntity::getSort, DatadevMenuNodeEntity::getNodeType);

        List<DatadevMenuNodeEntity> entityList = datadevMenuNodeService.list(lambdaQueryWrapper);

        if (CollectionUtils.isNotEmpty(entityList)) {
            return Collections.EMPTY_LIST;
        }

        Set<Long> includeTaskFlowDirIds = new HashSet<>();

        final Map<Long, Long> idPidMap =
            entityList.stream().collect(Collectors.toMap(x -> x.getId(), x -> x.getPid()));

        entityList.stream()
            .filter(
                node ->
                    Objects.equals(
                        node.getNodeType(), OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType()))
            .forEach(
                node -> {
                    Long nodeId = node.getId();
                    Long dirId = idPidMap.get(nodeId);
                    while (!dirId.equals(OfflineDevTopNodeIdEnum.TASK_FLOW.getId())) {
                        includeTaskFlowDirIds.add(dirId);
                        dirId = idPidMap.get(dirId);
                    }
                });

        // 排除无工作流节点的目录
        final List<OfflineDevTaskFlowDomain> domainList =
            entityList.stream()
                .filter(
                    node ->
                        Objects.equals(
                            node.getNodeType(), OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType())
                            || includeTaskFlowDirIds.contains(node.getId()))
                .map(
                    node -> {
                        OfflineDevTaskFlowDomain domain = new OfflineDevTaskFlowDomain();
                        domain.setId(node.getId());
                        domain.setNodeType(node.getNodeType());
                        domain.setName(node.getName());
                        domain.setPid(node.getPid());
                        return domain;
                    })
                .collect(Collectors.toList());

        return domainList;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean saveNodeCoord(OfflineDevSaveNodeCoordParam param, Long projectId, String userId) {
        final DatadevMenuNodeEntity node = datadevMenuNodeService.getById(param.getTaskFlowNodeId());
        if (null == node) {
            throw new BusinessException("节点不存在");
        }

        List<DatadevSubNodeEntity> subTaskList =
            datadevSubNodeService.getDatadevSubNodeList(param.getTaskFlowNodeId());

        final Map<Long, OfflineDevSaveNodeCoordParam.NodeCoord> nodeCoordMap =
            param.getCoordList().stream().collect(Collectors.toMap(x -> x.getNodeId(), x -> x));

        for (DatadevSubNodeEntity taskFlowSubTask : subTaskList) {
            final OfflineDevSaveNodeCoordParam.NodeCoord nodeCoord =
                nodeCoordMap.get(taskFlowSubTask.getNodeId());
            if (null == nodeCoord) {
                continue;
            }
            taskFlowSubTask.setCoordX(nodeCoord.getX());
            taskFlowSubTask.setCoordY(nodeCoord.getY());
        }

        datadevSubNodeService.updateBatchById(subTaskList);

        datadevOperationHistoryService.insertHistory(param.getTaskFlowNodeId(), projectId,
            "所有节点一键布局");
//        LogContext.put(LogKey.NAME.getKey(), node.getName());
//        LogContext.put(LogKey.ISUPDATE.getKey(), LogKey.ISUPDATE.getValue());
        return true;
    }

    @Override
    public DatadevMenuNodeEntity findByXxlJobId(Integer xxlJobId) {
        LambdaQueryWrapper<DatadevMenuNodeEntity> lambdaQueryWrapper =
            new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                .eq(DatadevMenuNodeEntity::getXxlJobId, xxlJobId)
                .eq(DatadevMenuNodeEntity::getRowState, 1);
        final DatadevMenuNodeEntity node = datadevMenuNodeService.getOne(lambdaQueryWrapper, false);
        return node;
    }

    @Override
    public DatadevMenuNodeEntity findById(Long id) {
        return datadevMenuNodeService.getById(id);
    }

    @Override
    public DatadevMenuNodeEntity findByIdAndProjectId(Long id, Long projectId) {
        final DatadevMenuNodeEntity node = datadevMenuNodeService.getById(id);
        if (null == node) {
            throw new BusinessException("节点不存在");
        }
        return node;
    }

    @Override
    public List<DatadevMenuNodeEntity> findByIdList(List<Long> idList) {
        if (CollectionUtils.isEmpty(idList)) {
            return Collections.EMPTY_LIST;
        }
        return datadevMenuNodeService.listByIds(idList);
    }

    @Override
    public List<OfflineDevResourceDomain> listResourceTask(Long id, Long projectId) {

        List<DatadevResourceEntity> resourceConfig = datadevResourceService.getResourceByNodeId(id);

        List<OfflineDevResourceDomain> resourceDomains =
            ConvertUtil.copyProperties(resourceConfig, OfflineDevResourceDomain.class);

        resourceDomains.forEach(e -> e.setNodeId(id));
        return resourceDomains;
    }

    //    @BizDataIndexAnnotation(
//        type = BizDataIndexTypeEnum.OFFLINE_TASKFLOW,
//        dataIdFieldName = "id",
//        dataIdParamIndex = 0,
//        operateType = BizDataIndexAnnotation.OperateType.DELETE)
    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public boolean dropResourceByIdAndName(Long id, String name, Long projectId) {
        List<DatadevResourceEntity> resourceConfig = datadevResourceService.getResourceByNodeId(id);
        for (DatadevResourceEntity resource : resourceConfig) {
            if (resource.getFileName().equalsIgnoreCase(name) && resource.getIsActive().equals(1)) {
                throw new BusinessException("最新的资源不能被删除");
            }
        }

        Iterator<DatadevResourceEntity> iterator = resourceConfig.iterator();
        while (iterator.hasNext()) {
            DatadevResourceEntity next = iterator.next();
            if (StringUtils.isNotBlank(next.getFileName()) && next.getFileName().equals(name)) {
                iterator.remove();
                datadevResourceService.removeById(next.getId());
                break;
            }
        }
        return true;
    }

    @Override
    public OfflineDevResourceDomain getResourceDetail(Long id, Long projectId) {

        DatadevMenuNodeEntity menuNode = datadevMenuNodeService.getById(id);

        List<DatadevResourceEntity> resourceConfig = datadevResourceService.getResourceByNodeId(id);
        DatadevResourceEntity resource =
            resourceConfig.stream().filter(e -> e.getIsActive().equals(1)).findFirst().get();
        OfflineDevResourceDomain devResourceDomain =
            ConvertUtil.copyProperties(resource, OfflineDevResourceDomain.class);
        devResourceDomain.setName(menuNode.getName());
        return devResourceDomain;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public void saveOrUpdateOfflineSyncTask(OfflineDevSyncDomain offlineSyncDomain) {
        Long projectId = ServletUtils.getProjectId();
        String userId = UserContext.getUserId();
        Long nodeId = offlineSyncDomain.getNodeId();
        DatadevMenuNodeEntity entity = datadevMenuNodeService.getById(nodeId);

        final boolean hasSameNode =
            datadevMenuNodeService.hasSameNode(
                nodeId, entity.getPid(), offlineSyncDomain.getNodeName(), projectId);
        if (hasSameNode) {
            throw new BusinessException("同名节点已存在");
        }
        List<PartitionField> partitionFields = offlineSyncDomain.getParitions();
        List<FieldMapping> fieldMappings = offlineSyncDomain.getFieldMappings();
        Integer pgText = offlineSyncDomain.getPgText();
        if (Objects.nonNull(pgText) && pgText.equals(GP_TEXT)) {
            int fileNum = 0;
            // 数量匹配
            OfflineTaskExistTableMappingQueryDomain r1 = new OfflineTaskExistTableMappingQueryDomain();
            r1.setDsId(offlineSyncDomain.getTargetDsId());
            r1.setTables(Lists.newArrayList(offlineSyncDomain.getTargetTable()));
            OfflineTaskExistTableMappingDomain existTablesMappings =
                offlineSyncService.getExistTablesMappings(r1);
            if (Objects.nonNull(existTablesMappings)) {
                List<Table> tables = existTablesMappings.getTables();
                if (org.apache.commons.collections.CollectionUtils.isNotEmpty(tables)) {
                    Table table = tables.get(0);
                    List<Field> fields = table.getFields();
                    fileNum = fields.size();
                }
            }
            // 字段匹配
            fieldMappings.forEach(
                e -> {
                    if (!e.getSourceField()
                        .toLowerCase(Locale.ROOT)
                        .equals(e.getTargetField().toLowerCase(Locale.ROOT))) {
                        throw new BusinessException("来源和目标字段映射的名称不相同，请重新编辑！");
                    }
                });
            if (fieldMappings.size() != fileNum) {
                throw new BusinessException("目标数据源的所有字段均需要连线，请重新编辑");
            }
        }
        // 把旧的分区信息移除
        datadevSyncPartitionService.remove(
            new LambdaQueryWrapper<DatatdevSyncPartitionEntity>()
                .eq(DatatdevSyncPartitionEntity::getNodeId, nodeId));
        // 把旧的mapping移除掉
        datadevSyncFieldMappingService.remove(
            new LambdaQueryWrapper<DatatdevSyncFieldMappingEntity>()
                .eq(DatatdevSyncFieldMappingEntity::getNodeId, nodeId));
        datadevOfflineSyncService.remove(
            new LambdaQueryWrapper<DatadevOfflineSyncEntity>()
                .eq(DatadevOfflineSyncEntity::getNodeId, nodeId));

        if (CollectionUtils.isNotEmpty(partitionFields)) {
            List<DatatdevSyncPartitionEntity> collect =
                partitionFields.stream()
                    .map(
                        e -> {
                            DatatdevSyncPartitionEntity convert =
                                Convert.convert(DatatdevSyncPartitionEntity.class, e);
                            convert.setNodeId(nodeId);
                            return convert;
                        })
                    .collect(Collectors.toList());
            datadevSyncPartitionService.saveBatch(collect);
        }

        if (CollectionUtils.isNotEmpty(fieldMappings)) {
            List<DatatdevSyncFieldMappingEntity> collect =
                fieldMappings.stream()
                    .map(
                        e -> {
                            DatatdevSyncFieldMappingEntity convert =
                                Convert.convert(DatatdevSyncFieldMappingEntity.class, e);
                            convert.setNodeId(nodeId);
                            return convert;
                        })
                    .collect(Collectors.toList());
            datadevSyncFieldMappingService.saveBatch(collect);
        }

        String targetTable = offlineSyncDomain.getTargetTable();
        if (!targetTable.contains(".")) {
            Long targetDsId = offlineSyncDomain.getTargetDsId();
            DataSourceDetailDomain targetDatasource =
                dataSourceService.getDataSourceDetailById(targetDsId, false);
            if (targetDatasource.getDsType().equals(DSType.Sap)) {
                targetTable = targetDatasource.getRdbmsConfig().getSchema() + "." + targetTable;
                offlineSyncDomain.setTargetTable(targetTable);
            }
        }

        String sourceTable = offlineSyncDomain.getSourceTable();
        if (!sourceTable.contains(".")) {
            Long sourceDsId = offlineSyncDomain.getSourceDsId();
            DataSourceDetailDomain sourceDatasource =
                dataSourceService.getDataSourceDetailById(sourceDsId, false);
            if (sourceDatasource.getDsType().equals(DSType.Sap)) {
                sourceTable = sourceDatasource.getRdbmsConfig().getSchema() + "." + sourceTable;
                offlineSyncDomain.setSourceTable(sourceTable);
            }
        }

        datadevOperationHistoryService.insertHistory(
            offlineSyncDomain.getId(),
            projectId,
            String.format("修改任务节点:[%s]的配置", offlineSyncDomain.getNodeName()));

        DatadevOfflineSyncEntity convert =
            Convert.convert(DatadevOfflineSyncEntity.class, offlineSyncDomain);
        convert.setUpdateTime(new Date());
        convert.setCreateTime(new Date());
        convert.setCreatorId(userId);
        convert.setUpdaterId(userId);
        datadevOfflineSyncService.saveOrUpdate(convert);
    }

    @Override
    public OfflineDevSyncDomain getOfflineDevSyncDomainByNodeId(Long nodeId) {

        DatadevOfflineSyncEntity offlineDevSyncTask =
            datadevOfflineSyncService.getDatadevOfflineSyncNodeId(nodeId);
        OfflineDevSyncDomain offlineSyncDomain =
            Convert.convert(OfflineDevSyncDomain.class, offlineDevSyncTask);
        List<DatatdevSyncFieldMappingEntity> syncFieldMappings =
            datadevSyncFieldMappingService.getSyncFieldMappingByNodeId(nodeId);
        List<DatatdevSyncPartitionEntity> syncPartitionEntities =
            datadevSyncPartitionService.getSyncFieldMappingByNodeId(nodeId);

        List<PartitionField> partitionFieldList =
            Convert.toList(PartitionField.class, syncPartitionEntities);
        List<FieldMapping> fieldMappingList = Convert.toList(FieldMapping.class, syncFieldMappings);
        offlineSyncDomain.setParitions(partitionFieldList);
        offlineSyncDomain.setFieldMappings(fieldMappingList);

        DataSourceDetailDomain sourceDetail =
            dataSourceService.getDataSourceDetailById(offlineSyncDomain.getSourceDsId(), false);
        String sourceDSName =
            Optional.ofNullable(sourceDetail).map(s -> s.getDsName()).orElse("数据源已删除");
        DataSourceDetailDomain targetDetail =
            dataSourceService.getDataSourceDetailById(offlineSyncDomain.getTargetDsId(), false);
        String targetDSName =
            Optional.ofNullable(targetDetail).map(s -> s.getDsName()).orElse("数据源已删除");
        if (DSType.hasSchema(sourceDetail.getDsType())) {
            String sourceTable = offlineSyncDomain.getSourceTable();
            if (sourceTable.contains(".")) {
                String[] split = sourceTable.split("\\.", 2);
                offlineSyncDomain.setSourceSchema(split[0]);
            }
        }
        if (DSType.hasSchema(targetDetail.getDsType())) {
            String targetTable = offlineSyncDomain.getTargetTable();
            if (targetTable.contains(".")) {
                String[] split = targetTable.split("\\.", 2);
                offlineSyncDomain.setTargetSchema(split[0]);
            }
        }
        offlineSyncDomain.setSourceDsName(sourceDSName);
        offlineSyncDomain.setTargetDsName(targetDSName);
        return offlineSyncDomain;
    }

    @Override
    public List<OfflineDevResourceListDomain> getOfflineDevResourceList(Long projectId) {

        List<DatadevResourceEntity> list =
            datadevResourceService.list(
                new LambdaQueryWrapper<DatadevResourceEntity>()
                    .eq(DatadevResourceEntity::getProjectId, projectId)
                    .eq(DatadevResourceEntity::getIsActive, 1));

        List<OfflineDevResourceListDomain> res = Lists.newArrayList();

        for (DatadevResourceEntity datadevResource : list) {
            DatadevMenuNodeEntity r1 = datadevMenuNodeService.getById(datadevResource.getNodeId());
            OfflineDevResourceListDomain resourceListDomain = new OfflineDevResourceListDomain();
            resourceListDomain.setConfigId(datadevResource.getNodeId());
            resourceListDomain.setName(r1.getName());
            res.add(resourceListDomain);
        }
        return res;
    }

    //    @BizDataIndexAnnotation(
//        type = BizDataIndexTypeEnum.API_HTTP_COLLECT,
//        dataIdFieldName = "nodeId",
//        dataIdParamIndex = 0,
//        operateType = BizDataIndexAnnotation.OperateType.SAVE)
    @Override
    public void saveOrUpdateOfflineApiTask(OfflineDevApiDomain offlineDevApiDomain) {
        Long projectId = ServletUtils.getProjectId();
        String userId = UserContext.getUserId();
        DatadevMenuNodeEntity entity = datadevMenuNodeService.getById(offlineDevApiDomain.getNodeId());

        final boolean hasSameNode =
            datadevMenuNodeService.hasSameNode(
                offlineDevApiDomain.getNodeId(),
                entity.getPid(),
                offlineDevApiDomain.getNodeName(),
                projectId);
        if (hasSameNode) {
            throw new BusinessException("同名节点已存在");
        }

        DatadevOfflineApiEntity datadevOfflineApiEntity =
            Convert.convert(DatadevOfflineApiEntity.class, offlineDevApiDomain);

        TaskDomain.PreviewExpressRequest previewExpressRequest =
            offlineDevApiDomain.getPreviewExpressRequest();
        OfflineDevApiPreviewEntity offlineDevApiPreviewEntity =
            ConvertUtil.copyProperties(previewExpressRequest, OfflineDevApiPreviewEntity.class);
        if (CollectionUtils.isNotEmpty(previewExpressRequest.getExpressionParamList())) {
            offlineDevApiPreviewEntity.setExpressionParamList(
                ConvertUtil.copyProperties(
                    previewExpressRequest.getExpressionParamList(),
                    OfflineDevApiPreviewEntity.ExpressionParam.class));
        }
        datadevOfflineApiEntity.setOfflineDevApiPreviewEntity(offlineDevApiPreviewEntity);

        // task config
        TaskConfigDomain taskConfigDomain = offlineDevApiDomain.getTaskConfigDomain();
        DatadevOfflineApiEntity.TaskConfigRequest entityTaskConfig =
            ConvertUtil.copyProperties(
                taskConfigDomain, DatadevOfflineApiEntity.TaskConfigRequest.class);

        if (CollectionUtils.isNotEmpty(taskConfigDomain.getTargetFields())) {
            entityTaskConfig.setTargetFields(
                ConvertUtil.copyProperties(
                    taskConfigDomain.getTargetFields(),
                    DatadevOfflineApiEntity.TaskConfigRequest.TargetField.class));
        }

        if (CollectionUtils.isNotEmpty(taskConfigDomain.getFieldMappings())) {
            entityTaskConfig.setFieldMappings(
                ConvertUtil.copyProperties(
                    taskConfigDomain.getFieldMappings(),
                    DatadevOfflineApiEntity.TaskConfigRequest.FieldMapping.class));
        }

        if (CollectionUtils.isNotEmpty(taskConfigDomain.getPartitions())) {
            entityTaskConfig.setPartitions(
                ConvertUtil.copyProperties(
                    taskConfigDomain.getPartitions(),
                    DatadevOfflineApiEntity.TaskConfigRequest.PartitionField.class));
        }

        datadevOfflineApiEntity.setTaskConfigRequest(entityTaskConfig);

        if (Objects.equals(
            OfflineDevApiNodeTypeEnum.WS.getNodeDesc(), datadevOfflineApiEntity.getApiNodeType())) {
            TestApiWsParam offlineDevApiWsParam = offlineDevApiDomain.getOfflineDevApiWsParam();
            OfflineDevApiWsEntity offlineDevApiWsEntity =
                ConvertUtil.copyProperties(offlineDevApiWsParam, OfflineDevApiWsEntity.class);
            // ws auth
            offlineDevApiWsEntity.setApiWsAuthParam(
                ConvertUtil.copyProperties(
                    offlineDevApiWsParam.getApiWsAuthParam(),
                    OfflineDevApiWsEntity.ApiHttpAuthParam.class));
            // ws body
            ApiWsRequestBodyParam apiWsRequestBodyParam = offlineDevApiWsParam.getApiWsRequestBodyParam();
            if (null != apiWsRequestBodyParam) {
                OfflineDevApiWsEntity.ApiHttpRequestBodyParam entityBodyParam =
                    new OfflineDevApiWsEntity.ApiHttpRequestBodyParam();
                entityBodyParam.setContentType(apiWsRequestBodyParam.getContentType());
                entityBodyParam.setBody(apiWsRequestBodyParam.getBody());
                if (CollectionUtils.isNotEmpty(apiWsRequestBodyParam.getFormParams())) {
                    entityBodyParam.setFormParams(
                        ConvertUtil.copyProperties(
                            apiWsRequestBodyParam.getFormParams(),
                            OfflineDevApiWsEntity.ApiHttpRequestBodyParam.FormParam.class));
                }
                offlineDevApiWsEntity.setApiWsRequestBodyParam(entityBodyParam);
            }
            datadevOfflineApiEntity.setOfflineDevApiWsEntity(offlineDevApiWsEntity);
        } else {

            TestApiHttpParam offlineDevApiHttpParam = offlineDevApiDomain.getOfflineDevApiHttpParam();
            OfflineDevApiHttpEntity offlineDevApiHttpEntity =
                ConvertUtil.copyProperties(offlineDevApiHttpParam, OfflineDevApiHttpEntity.class);
            // http auth
            offlineDevApiHttpEntity.setApiHttpAuthParam(
                ConvertUtil.copyProperties(
                    offlineDevApiHttpParam.getApiHttpAuthParam(),
                    OfflineDevApiHttpEntity.ApiHttpAuthParam.class));

            // http body
            ApiHttpRequestBodyParam apiHttpRequestBodyParam =
                offlineDevApiHttpParam.getApiHttpRequestBodyParam();
            if (null != apiHttpRequestBodyParam) {
                OfflineDevApiHttpEntity.ApiHttpRequestBodyParam entityBodyParam =
                    new OfflineDevApiHttpEntity.ApiHttpRequestBodyParam();
                entityBodyParam.setContentType(apiHttpRequestBodyParam.getContentType());
                entityBodyParam.setBody(apiHttpRequestBodyParam.getBody());
                if (CollectionUtils.isNotEmpty(apiHttpRequestBodyParam.getFormParams())) {
                    entityBodyParam.setFormParams(
                        ConvertUtil.copyProperties(
                            apiHttpRequestBodyParam.getFormParams(),
                            OfflineDevApiHttpEntity.ApiHttpRequestBodyParam.FormParam.class));
                }
                offlineDevApiHttpEntity.setApiHttpRequestBodyParam(entityBodyParam);
            }
            datadevOfflineApiEntity.setOfflineDevApiHttpEntity(offlineDevApiHttpEntity);
        }

        datadevOperationHistoryService.insertHistory(
            offlineDevApiDomain.getNodeId(),
            projectId,
            String.format("修改任务节点:[%s]的配置", offlineDevApiDomain.getNodeName()));

        datadevOfflineApiService.remove(
            new LambdaQueryWrapper<DatadevOfflineApiEntity>()
                .eq(DatadevOfflineApiEntity::getNodeId, offlineDevApiDomain.getNodeId()));

        datadevOfflineApiEntity.setCreatorId(userId);
        datadevOfflineApiEntity.setUpdaterId(userId);
        datadevOfflineApiEntity.setCreateTime(new Date());
        datadevOfflineApiEntity.setUpdateTime(new Date());
        datadevOfflineApiService.save(datadevOfflineApiEntity);
//        LogContext.put(LogKey.NAME.getKey(), datadevOfflineApiEntity.getTaskName());
//        LogContext.put(LogKey.ISUPDATE.getKey(), LogKey.ISUPDATE.getValue());
    }

    /**
     * 按nodeId查接口采集任务
     *
     * @param nodeId
     * @return
     */
    @Override
    public OfflineDevApiDomain getOfflineDevApiDomainByNodeId(Long nodeId) {
        return datadevOfflineApiService.getApiTaskByNodeId(nodeId);
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public boolean saveFunctionTask(OfflineDevFunctionSaveParam offlineDevFunctionSaveParam) {

        Long nodeId = offlineDevFunctionSaveParam.getNodeId();
        Function function = ConvertUtil.copyProperties(offlineDevFunctionSaveParam, Function.class);
        final Integer maxSort =
            datadevMenuNodeService.getMaxSortByPid(offlineDevFunctionSaveParam.getPid());
        // 编辑
        if (Objects.nonNull(nodeId)) {
            DatadevMenuNodeEntity offlineDevMenuNodeEntity = datadevMenuNodeService.getById(nodeId);
            //      OfflineDevTaskConfigEntity offlineDevTaskConfig =
            //          offlineDevTaskConfigService.getByNodeId(nodeId);
            //      offlineDevTaskConfig.setFunctionConfig(function);
            //      offlineDevTaskConfig.setUpdateTime(new Date());
            //      offlineDevTaskConfig.setUpdaterId(offlineDevFunctionSaveParam.getUserId());
            //      offlineDevTaskConfigService.updateById(offlineDevTaskConfig);
            final boolean hasSameNode =
                datadevMenuNodeService.hasSameNode(
                    nodeId,
                    offlineDevFunctionSaveParam.getPid(),
                    offlineDevFunctionSaveParam.getNodeName(),
                    offlineDevFunctionSaveParam.getProjectId());
            if (hasSameNode) {
                throw new BusinessException("同名节点已存在");
            }
            offlineDevMenuNodeEntity.setPid(offlineDevFunctionSaveParam.getPid());
            if (!offlineDevMenuNodeEntity.getPid().equals(offlineDevFunctionSaveParam.getPid())) {
                offlineDevMenuNodeEntity.setSort(maxSort + 1);
            }
            offlineDevMenuNodeEntity.setName(offlineDevFunctionSaveParam.getFunctionName());
            offlineDevMenuNodeEntity.setUpdateTime(new Date());
            datadevMenuNodeService.updateById(offlineDevMenuNodeEntity);
        } else {
            final boolean hasSameNode =
                datadevMenuNodeService.hasSameNode(
                    null,
                    offlineDevFunctionSaveParam.getPid(),
                    offlineDevFunctionSaveParam.getNodeName(),
                    offlineDevFunctionSaveParam.getProjectId());
            if (hasSameNode) {
                throw new BusinessException("同名节点已存在");
            }
            DatadevMenuNodeEntity offlineDevMenuNodeEntity = new DatadevMenuNodeEntity();
            offlineDevMenuNodeEntity.setProjectId(offlineDevFunctionSaveParam.getProjectId());
            offlineDevMenuNodeEntity.setName(offlineDevFunctionSaveParam.getFunctionName());
            offlineDevMenuNodeEntity.setSort(maxSort + 1);
            offlineDevMenuNodeEntity.setPid(offlineDevFunctionSaveParam.getPid());
            offlineDevMenuNodeEntity.setCreatorId(offlineDevFunctionSaveParam.getUserId());
            offlineDevMenuNodeEntity.setCreateTime(new Date());
            offlineDevMenuNodeEntity.setRowState(1);
            offlineDevMenuNodeEntity.setNodeType(OfflineDevNodeTypeEnum.FUNCTION.getNodeType());
            datadevMenuNodeService.save(offlineDevMenuNodeEntity);

            DatadevFunctionEntity convert = Convert.convert(DatadevFunctionEntity.class, function);
            convert.setNodeId(offlineDevMenuNodeEntity.getId());
            convert.setResourceId(function.getResourceNodeId());
            convert.setProjectId(ServletUtils.getProjectId());
            datadevFunctionService.save(convert);
        }

        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public boolean saveRule(
        OfflineDevDataQualityRuleParam dataqualityRule, Long projectId, UserDomain userDomain) {

        DatadevDataqualityEntity datadevDataqualityEntity =
            ConvertUtil.copyProperties(dataqualityRule, DatadevDataqualityEntity.class);

        List<DatadevDataqualityEntity> list =
            datadevDataqualityService.list(
                new LambdaQueryWrapper<DatadevDataqualityEntity>()
                    .eq(DatadevDataqualityEntity::getProjectId, projectId)
                    .eq(DatadevDataqualityEntity::getNodeId, dataqualityRule.getNodeId()));

        if (CollectionUtils.isNotEmpty(list)) {
            // update
            if (list.size() >= 2) {
                throw new BusinessException("一个nodeId对应一个数据质量任务");
            } else {
                Long id = list.get(0).getId();
                datadevDataqualityEntity.setId(id);
                datadevDataqualityEntity.setProjectId(projectId);
                datadevDataqualityService.updateById(datadevDataqualityEntity);
            }

        } else {
            // save
            datadevDataqualityEntity.setProjectId(projectId);
            datadevDataqualityService.save(datadevDataqualityEntity);
        }

        return true;
    }

    @Override
    public OfflineDevDataQualityDetailDomain getRule(Long nodeId) {

        DatadevDataqualityEntity dataqualityTaskConfig =
            datadevDataqualityService.getOne(
                new LambdaQueryWrapper<DatadevDataqualityEntity>()
                    .eq(DatadevDataqualityEntity::getNodeId, nodeId));

        if (dataqualityTaskConfig == null) {
            return null;
        }

        OfflineDevDataQualityDetailDomain dataQualityDetailDomain =
            ConvertUtil.copyProperties(dataqualityTaskConfig, OfflineDevDataQualityDetailDomain.class);

        List<String> compareTypeList = new ArrayList<>();
        if (dataqualityTaskConfig.getRuleCompareType() != null) {
            String[] ruleCompareTypeList = dataqualityTaskConfig.getRuleCompareType().split(",");
            for (String ruleCompareType : ruleCompareTypeList) {
                DataQualityRuleCompareTypeEnum compareType =
                    DataQualityRuleCompareTypeEnum.getByCompareType(Integer.valueOf(ruleCompareType));
                compareTypeList.add(compareType == null ? null : compareType.getDesc());
            }
        }
        dataQualityDetailDomain.setRuleCompareTypeName(
            StringUtils.join(compareTypeList.toArray(), ","));

        // 判断是否是自定义规则
        if (dataqualityTaskConfig.getRuleDiy() == 0) {
            DataQualityRuleTemplateDomain ruleTemplate =
                dataQualityManagerService.getRuleTemplateByTemplateId(
                    dataqualityTaskConfig.getTemplateId());

            if (null != ruleTemplate) {
                // 拼装规则名字
                dataQualityDetailDomain.setTemplateName(ruleTemplate.getName());

                if (StringUtils.isNotBlank(ruleTemplate.getConditionTemplate())) {
                    List<OfflineDevDataQualityDetailDomain.RuleCondition> ruleConditions =
                        JSON.parseArray(
                            ruleTemplate.getConditionTemplate(),
                            OfflineDevDataQualityDetailDomain.RuleCondition.class);
                    dataQualityDetailDomain.setRuleConditions(ruleConditions);
                }

                if (StringUtils.isNotBlank(ruleTemplate.getCompareTypeTemplate())) {
                    List<OfflineDevDataQualityDetailDomain.CompareType> compareTypes =
                        JSON.parseArray(
                            ruleTemplate.getCompareTypeTemplate(),
                            OfflineDevDataQualityDetailDomain.CompareType.class);
                    dataQualityDetailDomain.setCompareTypes(compareTypes);
                }

                if (StringUtils.isNotBlank(ruleTemplate.getExpectValueTemplate())) {
                    OfflineDevDataQualityDetailDomain.ExpectValue expectValue =
                        JSON.parseObject(
                            ruleTemplate.getExpectValueTemplate(),
                            OfflineDevDataQualityDetailDomain.ExpectValue.class);
                    dataQualityDetailDomain.setExpectValue(expectValue);
                }
            }
        }

        return dataQualityDetailDomain;
    }

    @Override
    public OfflineDevFunctionDomain getOfflineDevFunctionDomain(Long nodeId) {
        DatadevFunctionEntity functionConfig =
            datadevFunctionService.getDatadevFunctionByNodeId(nodeId);
        OfflineDevFunctionDomain offlineDevFunctionDomain =
            ConvertUtil.copyProperties(functionConfig, OfflineDevFunctionDomain.class);
        Long resourceNodeId = functionConfig.getResourceId();

        DatadevResourceEntity resource = datadevResourceService.getResourceIsActiveByNodeId(
            resourceNodeId);
        if (Objects.nonNull(resource)) {
            offlineDevFunctionDomain.setResourcePath(resource.getFileDownloadUrl());
        }

        offlineDevFunctionDomain.setResourceNodeId(resource.getNodeId());
        offlineDevFunctionDomain.setUpdateTime(functionConfig.getUpdateTime());
        offlineDevFunctionDomain.setCreateTime(functionConfig.getCreateTime());
        return offlineDevFunctionDomain;
    }

    //    @BizDataIndexAnnotation(
//        type = BizDataIndexTypeEnum.OFFLINE_JOB,
//        dataIdFieldName = "id",
//        dataIdParamIndex = 0,
//        operateType = BizDataIndexAnnotation.OperateType.SAVE)
    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public Boolean saveWorkTask(OfflineDevTaskSaveParam param, Long projectId, String userId) {

        if (null != param.getId()) {

            final boolean hasSameNode =
                datadevMenuNodeService.hasSameNode(
                    param.getId(), param.getPid(), param.getNodeName(), projectId);
            if (hasSameNode) {
                throw new BusinessException("同名节点已存在");
            }

            final Integer maxSort = datadevMenuNodeService.getMaxSortByPid(param.getPid());
            final DatadevMenuNodeEntity nodeEntity = datadevMenuNodeService.getById(param.getId());
            if (null == nodeEntity) {
                throw new BusinessException("节点不存在");
            }
            nodeEntity.setPid(param.getPid());
            nodeEntity.setName(param.getNodeName());
            nodeEntity.setRemark(param.getRemark());
            nodeEntity.setUpdaterId(userId);
            nodeEntity.setSort(maxSort + 1);
            datadevMenuNodeService.updateById(nodeEntity);
            //        LogContext.put(LogKey.ISUPDATE.getKey(), LogKey.ISUPDATE.getValue());
            //        LogContext.put(LogKey.NAME.getKey(), nodeEntity.getName());
            return true;
        }

        DatadevMenuNodeEntity entity = new DatadevMenuNodeEntity();
        entity.setId(param.getId());
        entity.setPid(param.getPid());
        entity.setName(param.getNodeName());
        entity.setRemark(param.getRemark());
        entity.setUpdaterId(userId);
        final boolean hasSameNode =
            datadevMenuNodeService.hasSameNode(null, param.getPid(), param.getNodeName(), projectId);
        if (hasSameNode) {
            throw new BusinessException("同名节点已存在");
        }

        final Integer maxSort = datadevMenuNodeService.getMaxSortByPid(param.getPid());

        entity.setCreatorId(userId);
        entity.setProjectId(projectId);
        entity.setNodeType(param.getNodeType());
        entity.setRowState(1);
        entity.setSort(maxSort + 1);
        datadevMenuNodeService.save(entity);
//        LogContext.put(LogKey.NAME.getKey(), entity.getName());

        param.setId(entity.getId());
        return true;
    }

    @Override
    public List<OfflineDevNodeDomain> getUntilNodeType(Long projectId, Integer nodeType) {

        LambdaQueryWrapper<DatadevMenuNodeEntity> eq =
            new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                .eq(DatadevMenuNodeEntity::getPid, 2)
                .eq(DatadevMenuNodeEntity::getProjectId, projectId)
                .eq(DatadevMenuNodeEntity::getRowState, 1);

        eq.orderByDesc(DatadevMenuNodeEntity::getSort);

        final List<DatadevMenuNodeEntity> nodeEntities = datadevMenuNodeService.list(eq);

        List<DatadevMenuNodeEntity> r2 = Lists.newArrayList();
        for (DatadevMenuNodeEntity nodeEntity : nodeEntities) {
            final OfflineDevNodeTypeEnum nodeTypeEnum =
                OfflineDevNodeTypeEnum.of(nodeEntity.getNodeType());
            if (nodeTypeEnum.isDir()) {
                r2.add(nodeEntity);
                getUntilByPid(nodeEntity.getId(), r2, projectId, nodeType);
            }
        }

        List<OfflineDevNodeDomain> allNodes =
            ConvertUtil.copyProperties(r2, OfflineDevNodeDomain.class);

        List<OfflineDevNodeDomain> r1 =
            allNodes.stream()
                .filter(e -> Objects.nonNull(e))
                .map(
                    e -> {
                        if (null == e.getSort()) {
                            e.setSort(0);
                        }
                        return e;
                    })
                .sorted(Comparator.comparing(offlineDevNodeDomain -> offlineDevNodeDomain.getSort()))
                .collect(Collectors.toList());
        return r1;
    }

    @Override
    public DatadevMenuNodeEntity getTaskFlowVnodeByTaskFlowId(Long taskFlowId, Long projectId) {
        DatadevMenuNodeEntity datadevMenuNodeEntity =
            datadevMenuNodeService.getDatadevMenuNodeByFlowId(
                taskFlowId, OfflineDevNodeTypeEnum.VNODE_START);
        return datadevMenuNodeEntity;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public boolean setTaskFlowScheduleStatus(Long taskFlowNodeId, Integer status) {
        DatadevMenuNodeEntity datadevMenuNodeEntity = datadevMenuNodeService.getById(taskFlowNodeId);

        Integer xxlJobId = datadevMenuNodeEntity.getXxlJobId();
        if (null == xxlJobId) {
            throw new BusinessException("请先配置调度周期，再发布、冻结任务");
        }

        if (status == 1) {
            xxlJobService.stop(xxlJobId);
        } else {
            xxlJobService.start(xxlJobId);
        }
        datadevMenuNodeEntity.setPublishTime(new Date());
        datadevMenuNodeEntity.setIsLocked(status);
        datadevMenuNodeService.updateById(datadevMenuNodeEntity);
//        LogContext.put(LogKey.NAME.getKey(), datadevMenuNodeEntity.getName());
//        LogContext.put(LogKey.ISUPDATE.getKey(), LogKey.ISUPDATE.getValue());
        return true;
    }

    @Override
    public List<OfflineDevNodeDomain> getAllTaskFlowList(Long projectId) {

        List<DatadevMenuNodeEntity> list =
            datadevMenuNodeService.list(
                new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                    .eq(DatadevMenuNodeEntity::getRowState, 1)
                    .eq(DatadevMenuNodeEntity::getProjectId, projectId));

        List<OfflineDevNodeDomain> offlineDevNodeDomains =
            ConvertUtil.copyProperties(list, OfflineDevNodeDomain.class);
        return offlineDevNodeDomains;
    }

    @Override
    public List<OfflineDevNodeDomain> getAllPublishedTaskFlowList(Long projectId) {

        List<DatadevMenuNodeEntity> list1 =
            datadevMenuNodeService.list(
                new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                    .eq(DatadevMenuNodeEntity::getProjectId, projectId)
                    .eq(DatadevMenuNodeEntity::getIsLocked, 0)
                    .eq(DatadevMenuNodeEntity::getRowState, 1)
                    .eq(
                        DatadevMenuNodeEntity::getNodeType,
                        OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType()));

        List<OfflineDevNodeDomain> offlineDevNodeDomains =
            ConvertUtil.copyProperties(list1, OfflineDevNodeDomain.class);
        return offlineDevNodeDomains;
    }

    @Override
    public Long getIdByNamePid(String name, Long pid) {
        Long projectId = ServletUtils.getProjectId();
        DatadevMenuNodeEntity one =
            datadevMenuNodeService.getOne(
                new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                    .eq(DatadevMenuNodeEntity::getName, name)
                    .eq(DatadevMenuNodeEntity::getPid, pid)
                    .eq(DatadevMenuNodeEntity::getRowState, 1)
                    .eq(DatadevMenuNodeEntity::getProjectId, projectId));
        if (one == null) {
            return null;
        }
        return one.getId();
    }

    @Override
    public List<OfflineDevNodeDomain> getNodeListByIds(List<Long> idList, Long projectId) {
        LambdaQueryWrapper<DatadevMenuNodeEntity> lambdaQueryWrapper =
            new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                .eq(DatadevMenuNodeEntity::getProjectId, projectId);
        if (CollectionUtils.isNotEmpty(idList)) {
            lambdaQueryWrapper.in(DatadevMenuNodeEntity::getId, idList);
        }
        List<DatadevMenuNodeEntity> one = datadevMenuNodeService.list(lambdaQueryWrapper);
        List<OfflineDevNodeDomain> offlineDevNodeDomains =
            ConvertUtil.copyProperties(one, OfflineDevNodeDomain.class);
        return offlineDevNodeDomains;
    }

    @Override
    public Long getParentTypeByNodeId(Long nodeId) {

        DatadevMenuNodeEntity menuNode = datadevMenuNodeService.getById(nodeId);
        if (menuNode.getPid().equals(Long.valueOf(OfflineDevTopNodeIdEnum.TASK.getId()))
            || menuNode.getPid().equals(Long.valueOf(OfflineDevTopNodeIdEnum.TASK_FLOW.getId()))) {
            return menuNode.getPid();
        }
        return getParentTypeByNodeId(menuNode.getPid());
    }

    @Override
    public OfflineDevToolsDomain getDevToolList(Long projectId) {

        List<Node> offlineSync =
            Lists.newArrayList(
                new Node(
                    OfflineDevNodeTypeEnum.OFFLINE.getNodeType(),
                    OfflineDevNodeTypeEnum.OFFLINE.getNodeDesc()),
                new Node(
                    OfflineDevNodeTypeEnum.API.getNodeType(),
                    OfflineDevNodeTypeEnum.API.getNodeDesc()));

        List<Node> datadev =
            Lists.newArrayList(
                new Node(
                    OfflineDevNodeTypeEnum.SQL.getNodeType(), OfflineDevNodeTypeEnum.SQL.getNodeDesc()),
                new Node(
                    OfflineDevNodeTypeEnum.SHELL.getNodeType(),
                    OfflineDevNodeTypeEnum.SHELL.getNodeDesc()),
                new Node(
                    OfflineDevNodeTypeEnum.PYTHON2.getNodeType(),
                    OfflineDevNodeTypeEnum.PYTHON2.getNodeDesc()),
                new Node(
                    OfflineDevNodeTypeEnum.PYTHON3.getNodeType(),
                    OfflineDevNodeTypeEnum.PYTHON3.getNodeDesc()));

        List<Node> dataQuality =
            Lists.newArrayList(
                new Node(
                    OfflineDevNodeTypeEnum.DATAQUALITY.getNodeType(),
                    OfflineDevNodeTypeEnum.DATAQUALITY.getNodeDesc()));

        ProjectEntity project = projectService.getProjectById(projectId);
        StorageEntity storage = storageService.getById(project.getStorageId());

        if (Objects.nonNull(storage) && storage.getStorageType().equals(DSType.Hive.name())) {
            datadev.add(
                new Node(
                    OfflineDevNodeTypeEnum.MR.getNodeType(), OfflineDevNodeTypeEnum.MR.getNodeDesc()));
            datadev.add(
                new Node(
                    OfflineDevNodeTypeEnum.PYSPARK.getNodeType(),
                    OfflineDevNodeTypeEnum.PYSPARK.getNodeDesc()));
            datadev.add(
                new Node(
                    OfflineDevNodeTypeEnum.IMPALA.getNodeType(),
                    OfflineDevNodeTypeEnum.IMPALA.getNodeDesc()));
        }
        OfflineDevToolsDomain offlineDevToolsDomain = new OfflineDevToolsDomain();
        offlineDevToolsDomain.setOfflineSync(offlineSync);
        offlineDevToolsDomain.setDatadev(datadev);
        offlineDevToolsDomain.setDataQuality(dataQuality);
        return offlineDevToolsDomain;
    }

    @Override
    public DatadevMenuNodeEntity getVnodeByTaskFlowNodeId(Long taskFlowNodeId) {
        DatadevMenuNodeEntity datadevMenuNodeByFlowId =
            datadevMenuNodeService.getDatadevMenuNodeByFlowId(
                taskFlowNodeId, OfflineDevNodeTypeEnum.VNODE_START);
        return datadevMenuNodeByFlowId;
    }

    @Override
    public List<DatadevMenuNodeEntity> getListByPage(
        Long projectId, Integer pageSize, Integer pageNum) {
        Page<DatadevMenuNodeEntity> page =
            datadevMenuNodeService.page(
                new Page<>(pageNum, pageSize),
                new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                    .eq(DatadevMenuNodeEntity::getProjectId, projectId)
                    .eq(DatadevMenuNodeEntity::getRowState, 1));
        return page.getRecords();
    }

    private void getUntilByPid(
        Long pid, List<DatadevMenuNodeEntity> nodes, Long projectId, Integer nodeType) {
        final List<DatadevMenuNodeEntity> nodeEntities = getNodeByPid(pid, projectId);
        for (DatadevMenuNodeEntity nodeEntity : nodeEntities) {
            final OfflineDevNodeTypeEnum nodeTypeEnum =
                OfflineDevNodeTypeEnum.of(nodeEntity.getNodeType());
            if (nodeTypeEnum.isUnitTaskPlus(nodeType) || nodeEntity.getNodeType().equals(nodeType)) {
                DatadevMenuNodeEntity offlineDevNodeDomain =
                    ConvertUtil.copyProperties(nodeEntity, DatadevMenuNodeEntity.class);
                nodes.add(offlineDevNodeDomain);
                getUntilByPid(nodeEntity.getId(), nodes, projectId, nodeType);
            }
        }
    }

    /**
     * 循环连线校验
     *
     * @param dependencies
     * @param upNodeId
     * @param downNodeId
     * @return
     */
    private Boolean checkCycleDependency(
        List<DatadevSubNodeDependencyEntity> dependencies, Long upNodeId, Long downNodeId) {

        List<DatadevSubNodeDependencyEntity> downTasks =
            dependencies.stream()
                .filter(d -> Objects.equals(d.getUpNodeId(), downNodeId))
                .collect(Collectors.toList());

        // 如果下游任务的下游节点有上游任务，则判为循环
        if (CollectionUtils.isNotEmpty(downTasks)) {
            long count =
                downTasks.stream().filter(d -> Objects.equals(d.getDownNodeId(), upNodeId)).count();
            if (count > 0) {
                return true;
            }
            for (DatadevSubNodeDependencyEntity downTask : downTasks) {
                if (checkCycleDependency(dependencies, upNodeId, downTask.getDownNodeId())) {
                    return true;
                }
            }
        }
        return false;
    }

    private DatadevMenuNodeEntity getByTaskFlowNodeIdAndDefaultNodeType(
        Long nodeId, Integer nodeType) {

        DatadevMenuNodeEntity node =
            datadevMenuNodeService.getOne(
                new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                    .eq(DatadevMenuNodeEntity::getPid, nodeId)
                    .eq(DatadevMenuNodeEntity::getIsSystem, 1)
                    .eq(DatadevMenuNodeEntity::getNodeType, nodeType));
        return node;
    }

    private void getChildDirs(
        Long pid, List<DatadevMenuNodeEntity> nodes, OfflineDevDirTreeDomain domain) {
        List<OfflineDevDirTreeDomain> childs = new ArrayList<>();

        nodes.stream()
            .filter(
                node ->
                    (Objects.equals(node.getPid(), pid)
                        && ((null != node.getNodeType())
                        && OfflineDevNodeTypeEnum.of(node.getNodeType()).isDir())))
            .forEach(
                node -> {
                    OfflineDevDirTreeDomain child = new OfflineDevDirTreeDomain();
                    child.setId(node.getId());
                    child.setName(node.getName());
                    child.setPid(node.getPid());
                    child.setNodeType(node.getNodeType());
                    child.setSort(node.getSort());
                    childs.add(child);

                    getChildDirs(node.getId(), nodes, child);
                });

        domain.setChilds(childs);
    }

    /**
     * 将子任务移动到其他工作流程中
     *
     * @param taskNodeId
     * @param newTaskFlowNodeId
     */
    private void setTaskNodeToNewTaskFlow(
        Long taskNodeId, Long oldTaskFlowNodeId, Long newTaskFlowNodeId) {
        // 删除连线，删除工作流节点列表
        List<DatadevSubNodeEntity> subNodeList =
            datadevSubNodeService.getDatadevSubNodeList(oldTaskFlowNodeId);

        List<DatadevSubNodeDependencyEntity> subTaskDependencyList =
            datadevSubNodeDependencyService.getDatadevSubNodeDepenDencyList(oldTaskFlowNodeId);

        if (CollectionUtils.isNotEmpty(subTaskDependencyList)) {
            final Iterator<DatadevSubNodeDependencyEntity> dependencyIterator =
                subTaskDependencyList.iterator();
            while (dependencyIterator.hasNext()) {
                final DatadevSubNodeDependencyEntity next = dependencyIterator.next();
                if (next.getDownNodeId().equals(taskNodeId) || next.getUpNodeId().equals(taskNodeId)) {
                    datadevSubNodeDependencyService.removeById(next.getId());
                    dependencyIterator.remove();
                }
            }
        }
        final Iterator<DatadevSubNodeEntity> subTaskIterator = subNodeList.iterator();
        DatadevSubNodeEntity taskFlowSubTask = null;
        while (subTaskIterator.hasNext()) {
            final DatadevSubNodeEntity next = subTaskIterator.next();
            if (Objects.isNull(next.getNodeId())) {
                continue;
            }
            if (next.getNodeId().equals(taskNodeId)) {
                taskFlowSubTask = next;
                subTaskIterator.remove();
                datadevSubNodeService.removeById(next.getId());
            }
        }

        // 将节点放置到新的工作流里面
        taskFlowSubTask.setId(null);
        taskFlowSubTask.setTaskFlowId(newTaskFlowNodeId);
        datadevSubNodeService.save(taskFlowSubTask);
    }

    /**
     * 获取顶级是系统级的node
     *
     * @param id
     * @return
     */
    private DatadevMenuNodeEntity getTopTaskFlowDefaultDir(Long id) {
        if (OfflineDevTopNodeIdEnum.getTopNodeIds().contains(id)) {
            return null;
        }

        DatadevMenuNodeEntity node = datadevMenuNodeService.getById(id);

        OfflineDevNodeTypeEnum nodeTypeEnum = OfflineDevNodeTypeEnum.of(node.getNodeType());

        while (!nodeTypeEnum.isTaskFlowDefaultDir()) {

            if (OfflineDevTopNodeIdEnum.getTopNodeIds().contains(node.getPid())) {
                return null;
            }
            node = datadevMenuNodeService.getById(node.getPid());
            nodeTypeEnum = OfflineDevNodeTypeEnum.of(node.getNodeType());
        }
        return node;
    }

    private DatadevMenuNodeEntity getTopNode(Long id) {
        if (TOP_NODE_ID.contains(id)) {
            OfflineDevTopNodeIdEnum of = OfflineDevTopNodeIdEnum.of(String.valueOf(id));
            DatadevMenuNodeEntity offlineDevMenuNodeEntity = new DatadevMenuNodeEntity();
            offlineDevMenuNodeEntity.setPid(Long.valueOf(of.getId()));
            offlineDevMenuNodeEntity.setName(of.getName());
            return offlineDevMenuNodeEntity;
        }
        DatadevMenuNodeEntity node = datadevMenuNodeService.getById(id);
        while (!DEFAULT_SYSTEM_ID.contains(node.getNodeType())
            && !TOP_NODE_ID.contains(node.getPid())) {
            node = datadevMenuNodeService.getById(node.getPid());
        }
        return node;
    }

    /**
     * 删除工作流
     *
     * @param includeChildTask
     */
    private void deleteTaskFlow(Long nodeId, boolean includeChildTask) {
        // 如果是物理删除，需要同时把工作流内的所有子节点删除
        final List<DatadevSubNodeEntity> subTaskList =
            datadevSubNodeService.list(
                new LambdaQueryWrapper<DatadevSubNodeEntity>()
                    .eq(DatadevSubNodeEntity::getTaskFlowId, nodeId));
        final List<DatadevSubNodeDependencyEntity> subTaskDependencyList =
            datadevSubNodeDependencyService.list(
                new LambdaQueryWrapper<DatadevSubNodeDependencyEntity>()
                    .eq(DatadevSubNodeDependencyEntity::getTaskFlowId, nodeId));
        if (!includeChildTask) {
            if (subTaskList.size() > 1) {
                throw new BusinessException("该业务流程内部还有未删除的节点，不允许删除");
            }
        }

        // 删除所有子节点
        for (DatadevSubNodeEntity subTask : subTaskList) {
            this.deleteNode(subTask.getNodeId(), subTask.getProjectId(), true, true, nodeId);
        }
        datadevMenuNodeService.removeById(nodeId);
        // todo 是否删除历史记录
    }

    /**
     * 删除目录
     *
     * @param id
     */
    private void deleteDir(Long id) {
        // todo
    }

    /**
     * 删除资源
     *
     * @param nodeId
     */
    private void deleteResource(Long nodeId) {
        datadevMenuNodeService.removeById(nodeId);
        datadevResourceService.remove(
            new LambdaQueryWrapper<DatadevResourceEntity>()
                .eq(DatadevResourceEntity::getNodeId, nodeId));
    }

    /**
     * 删除函数
     *
     * @param nodeId
     */
    private void deleteFunction(Long nodeId) {
        datadevMenuNodeService.removeById(nodeId);
        datadevFunctionService.remove(
            new LambdaQueryWrapper<DatadevFunctionEntity>()
                .eq(DatadevFunctionEntity::getNodeId, nodeId));
    }

    private void deleteTask(Long nodeId) {
        final DatadevMenuNodeEntity topTaskFlowDefaultDir = getTopTaskFlowDefaultDir(nodeId);

        if (null != topTaskFlowDefaultDir) {

            final List<DatadevSubNodeEntity> subTaskList =
                datadevSubNodeService.list(
                    new LambdaQueryWrapper<DatadevSubNodeEntity>()
                        .eq(DatadevSubNodeEntity::getTaskFlowId, topTaskFlowDefaultDir.getId()));
            final List<DatadevSubNodeDependencyEntity> subTaskDependencyList =
                datadevSubNodeDependencyService.list(
                    new LambdaQueryWrapper<DatadevSubNodeDependencyEntity>()
                        .eq(
                            DatadevSubNodeDependencyEntity::getTaskFlowId,
                            topTaskFlowDefaultDir.getId()));

            if (CollectionUtils.isNotEmpty(subTaskList)) {
                final Iterator<DatadevSubNodeEntity> iterator = subTaskList.iterator();
                while (iterator.hasNext()) {
                    DatadevSubNodeEntity next = iterator.next();
                    if (Objects.isNull(iterator.next().getNodeId())) {
                        continue;
                    }
                    if (next.getNodeId().equals(nodeId)) {
                        iterator.remove();
                        datadevSubNodeService.removeById(next.getId());
                    }
                }
            }

            if (CollectionUtils.isNotEmpty(subTaskDependencyList)) {
                final Iterator<DatadevSubNodeDependencyEntity> iterator = subTaskDependencyList.iterator();
                while (iterator.hasNext()) {
                    final DatadevSubNodeDependencyEntity next = iterator.next();
                    if (next.getUpNodeId().equals(nodeId) || next.getDownNodeId().equals(nodeId)) {
                        iterator.remove();
                        datadevSubNodeDependencyService.removeById(next.getId());
                    }
                }
            }
        }

        datadevMenuNodeService.removeById(nodeId);
    }

    private void getAllNodeByPid(Long pid, List<DatadevMenuNodeEntity> nodes, Long projectId) {
        final List<DatadevMenuNodeEntity> nodeEntities = getNodeByPid(pid, projectId);

        nodes.addAll(nodeEntities);

        for (DatadevMenuNodeEntity nodeEntity : nodeEntities) {
            final OfflineDevNodeTypeEnum nodeTypeEnum =
                OfflineDevNodeTypeEnum.of(nodeEntity.getNodeType());
            if (null != nodeTypeEnum && nodeTypeEnum.isDir()) {
                getAllNodeByPid(nodeEntity.getId(), nodes, projectId);
            }
        }
    }

    private void getAllNodeByPid(
        Long pid, List<DatadevMenuNodeEntity> nodes, Map<Long, List<DatadevMenuNodeEntity>> map) {
        final List<DatadevMenuNodeEntity> nodeEntities = map.get(pid);

        if (CollectionUtils.isEmpty(nodeEntities)) {
            return;
        }
        nodes.addAll(nodeEntities);
        for (DatadevMenuNodeEntity nodeEntity : nodeEntities) {
            final OfflineDevNodeTypeEnum nodeTypeEnum =
                OfflineDevNodeTypeEnum.of(nodeEntity.getNodeType());
            if (null != nodeTypeEnum && nodeTypeEnum.isDir()) {
                getAllNodeByPid(nodeEntity.getId(), nodes, map);
            }
        }
    }

    private void getAllSystemNodeByPid(Long pid, List<DatadevMenuNodeEntity> nodes, Long projectId) {
        final List<DatadevMenuNodeEntity> nodeEntities = getNodeByPid(pid, projectId);

        nodes.addAll(nodeEntities);

        for (DatadevMenuNodeEntity nodeEntity : nodeEntities) {
            getAllSystemNodeByPid(nodeEntity.getId(), nodes, projectId);
        }
    }

    /**
     * 找外部的所有的目录，不包括工作流系统目录
     *
     * @param pid
     * @param nodes
     * @param projectId
     */
    private void getAllNodeIncludeByPid(Long pid, List<DatadevMenuNodeEntity> nodes, Long projectId) {
        final List<DatadevMenuNodeEntity> nodeEntities = getNodeByPid(pid, projectId);
        List<DatadevMenuNodeEntity> filtered =
            nodeEntities.stream()
                .filter(e -> !DEFAULT_SYSTEM_ID.contains(e.getNodeType()))
                .collect(Collectors.toList());
        nodes.addAll(filtered);

        for (DatadevMenuNodeEntity nodeEntity : nodeEntities) {
            Integer nodeType = nodeEntity.getNodeType();
            if (!DEFAULT_SYSTEM_ID.contains(nodeType)) {
                getAllNodeIncludeByPid(nodeEntity.getId(), nodes, projectId);
            }
        }
    }

    //  private void getTaskFlowDirNodes(String pid, List<OfflineDevMenuNodeEntity> nodes) {
    //    final List<OfflineDevMenuNodeEntity> nodeEntities = getNodeByPid(pid,);
    //
    //    nodeEntities.stream()
    //        .filter(node -> OfflineDevNodeTypeEnum.of(node.getNodeType()).isNormalDir())
    //        .forEach(
    //            node -> {
    //              nodes.add(node);
    //              getTaskFlowDirNodes(node.getId(), nodes);
    //            });
    //  }

    private List<DatadevMenuNodeEntity> getNodeByPid(Long pid, Long projectId) {
        //    //    OfflineDevMenuNodeEntity query = new OfflineDevMenuNodeEntity();
        //    //    query.setPid(pid);
        //    //    query.setRowState(1);
        //    //    query.setProjectId(projectId);
        //
        //    final Sort sort = Sort.by(Sort.Direction.DESC, "sort");

        final List<DatadevMenuNodeEntity> nodeEntities =
            datadevMenuNodeService.list(
                new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                    .eq(DatadevMenuNodeEntity::getPid, pid)
                    .eq(DatadevMenuNodeEntity::getRowState, 1)
                    .eq(DatadevMenuNodeEntity::getProjectId, projectId)
                    .orderByDesc(DatadevMenuNodeEntity::getSort));

        return nodeEntities;
    }

    @Override
    public void saveXxlJobByTaskFlowId(Long nodeId) {

        DatadevVnodeEntity vnodeStartConfig = datadevVnodeSerivce.getDatadevVnodeByNodeId(nodeId);
        if (Objects.isNull(vnodeStartConfig)) {
            return;
        }
        DatadevMenuNodeEntity node = datadevMenuNodeService.getById(nodeId);
        DatadevMenuNodeEntity byId = datadevMenuNodeService.getById(node.getPid());
        Integer xxlJobId = saveXxlJob(nodeId);
        byId.setXxlJobId(xxlJobId);
        byId.setPublishTime(new Date());
        xxlJobService.stop(xxlJobId);
        datadevMenuNodeService.updateById(byId);
    }

    @Override
    public List<OfflineDevNodeDomain> getDatadevList(Long projectId, Long dirId) {

        List<OfflineDevNodeDomain> result = Lists.newArrayList();
        generateTaskFlowList(result, dirId);
        List<String> collect = result.stream().map(e -> e.getCreatorId()).collect(Collectors.toList());
        Map<String, String> mapByUserIds = userService.getMapByUserIds(collect);
        result.forEach(e -> e.setUsername(mapByUserIds.get(e.getCreatorId())));
        return result;
    }

    private void generateTaskFlowList(List<OfflineDevNodeDomain> result, Long dirId) {
        LambdaQueryWrapper<DatadevMenuNodeEntity> eq =
            new LambdaQueryWrapper<>(DatadevMenuNodeEntity.class)
                .eq(DatadevMenuNodeEntity::getPid, dirId);
        List<DatadevMenuNodeEntity> list = datadevMenuNodeService.list(eq);
        for (DatadevMenuNodeEntity datadevMenuNodeEntity : list) {
            if (datadevMenuNodeEntity
                .getNodeType()
                .equals(OfflineDevNodeTypeEnum.DIR_NORMAL.getNodeType())) {
                generateTaskFlowList(result, datadevMenuNodeEntity.getId());
            } else if (datadevMenuNodeEntity
                .getNodeType()
                .equals(OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType())) {
                OfflineDevNodeDomain convert =
                    Convert.convert(OfflineDevNodeDomain.class, datadevMenuNodeEntity);
                result.add(convert);
            }
        }
    }

    private Integer saveXxlJob(Long nodeId) {

        DatadevVnodeEntity vnodeStartConfig = datadevVnodeSerivce.getDatadevVnodeByNodeId(nodeId);

        DatadevMenuNodeEntity vnode = datadevMenuNodeService.getById(nodeId);
        DatadevMenuNodeEntity taskFlowNode = datadevMenuNodeService.getById(vnode.getPid());
        if (vnodeStartConfig == null) {
            return null;
        }

        final UserDomain userInfo = UserContext.getUserInfo();

        //    final XxlJobGroup xxlHufuDefaultGroup = commonService.getXxlHufuDefaultGroup();
        //    if (null == xxlHufuDefaultGroup) {
        //      log.warn("xxl-job中未找到相关的执行器，调度将不能被执行，请排查!!!!!!");
        //    }

        int triggerStatus = 0;
        // 工作流未锁定且配置了调度,未发布
        if (taskFlowNode.getIsLocked().equals(0) && null != vnodeStartConfig) {
            triggerStatus = 1;
        }

        boolean isInsert = true;
        XxlJobInfo jobInfo = new XxlJobInfo();
        if (Objects.nonNull(taskFlowNode.getXxlJobId())) {
            jobInfo.setId(taskFlowNode.getXxlJobId());
            isInsert = false;
        }
        // groupID
        jobInfo.setJobGroup(XxlGroupEnum.DATA_ENGINE_SYNCTASK_GROUP.getId());
        // 任务描述
        jobInfo.setJobDesc(taskFlowNode.getName());
        // 责任人
        jobInfo.setAuthor(userInfo.getUserName());
        // 邮箱
        jobInfo.setAlarmEmail(userInfo.getEmail());
        // 调度执行器类型
        jobInfo.setScheduleType(ScheduleTypeEnum.HUFU.name());
        // 调度周期
        jobInfo.setScheduleConf(JSONObject.toJSONString(vnodeStartConfig));

        // 过期策略： 忽略
        jobInfo.setMisfireStrategy(MisfireStrategyEnum.DO_NOTHING.name());
        // 路由策略：轮询
        jobInfo.setExecutorRouteStrategy(ExecutorRouteStrategyEnum.RANDOM.name());
        // 执行器
        jobInfo.setExecutorHandler("OfflineDevTask");
        // 默认阻塞策略： 丢弃后续调度
        jobInfo.setExecutorBlockStrategy(ExecutorBlockStrategyEnum.DISCARD_LATER.name());
        // 运行模式：Bean
        jobInfo.setGlueType(GlueTypeEnum.BEAN.name());
        jobInfo.setTriggerStatus(triggerStatus);
        jobInfo.setEffectFrom(vnodeStartConfig.getEffectiveDateStart());
        jobInfo.setEffectTo(vnodeStartConfig.getEffectiveDateEnd());

        if (isInsert) {
            final ReturnT<String> returnT = xxlJobService.add(jobInfo);

            final String content = returnT.getContent();

            return Integer.valueOf(content);
        }

        final ReturnT<String> returnT = xxlJobService.update(jobInfo);

        return vnode.getXxlJobId();
    }

    @Override
    public List<DatadevMenuNodeEntity> getWorkFlowByProjectId(Long projectId) {
        return datadevMenuNodeService.list(
            new LambdaQueryWrapper<DatadevMenuNodeEntity>()
                .eq(DatadevMenuNodeEntity::getProjectId, projectId)
                .in(
                    DatadevMenuNodeEntity::getNodeType,
                    OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType())
                .eq(DatadevMenuNodeEntity::getRowState, 1));
    }

    @Override
    public List<String> getNodeTableById(Long jobId) {
        List<DatadevSubNodeDependencyEntity> datadevSubNodeDepenDencyList =
            datadevSubNodeDependencyService.getDatadevSubNodeDepenDencyList(jobId);
        Set<Long> nodeIds = Sets.newHashSet();
        for (DatadevSubNodeDependencyEntity dependencyEntity : datadevSubNodeDepenDencyList) {
            nodeIds.add(dependencyEntity.getDownNodeId());
            nodeIds.add(dependencyEntity.getUpNodeId());
        }
        Set<String> res = Sets.newHashSet();
        for (Long id : nodeIds) {
            DatadevMenuNodeEntity menuNode = datadevMenuNodeService.getById(id);
            Integer nodeType = menuNode.getNodeType();
            if (nodeType.equals(OfflineDevNodeTypeEnum.SQL.getNodeType())) {
                DatadevScriptEntity datadevScriptByNodeId =
                    datadevScriptService.getDatadevScriptByNodeId(id);
                String script = datadevScriptByNodeId.getScript();
                String param = datadevScriptByNodeId.getParam();
                List<Param> paramList = JSONArray.parseArray(param, Param.class);

                if (CollectionUtils.isEmpty(paramList)) {
                    paramList = Lists.newArrayList();
                }
                DataSourceDetailDomain dataSourceDetailById =
                    dataSourceService.getDataSourceDetailByProjectId(menuNode.getProjectId(), false);
                final Map<String, String> stringMap =
                    paramList.stream()
                        .collect(Collectors.toMap(x -> x.getParamName(), x -> x.getParamValue()));
                script = VarParserUtil.process(script, stringMap, null);
                String storageType = dataSourceDetailById.getDsType();
                if (storageType.equalsIgnoreCase("sap")) {
                    storageType = "sapdb";
                } else if (storageType.equalsIgnoreCase("kingbase8")) {
                    storageType = "kingbase";
                } else if (storageType.equalsIgnoreCase("DM")) {
                    storageType = "dm";
                }
                DbType dbType = DbType.of(storageType.toLowerCase());
                if (dbType.equals(DbType.tidb)) {
                    dbType = DbType.mysql;
                }
                List<String> scripts = Lists.newArrayList();
                if (dbType.equals(DbType.sapdb)) {
                    scripts = SqlParserUtil.sapValidSplitSql(script, EDbVendor.dbvhana);
                } else {
                    scripts = SqlParserUtil.split(script, dbType);
                }
                for (String str : scripts) {
                    Map<String, List<String>> stringListMap = secretMgtService.parserSQLSecurity(str);
                    List<String> strings = stringListMap.get("other");
                    res.addAll(strings);
                }
            } else if (nodeType.equals(OfflineDevNodeTypeEnum.OFFLINE.getNodeType())) {
                OfflineSyncDomain syncTaskByNodeId = datadevOfflineSyncService.getSyncTaskByNodeId(id);
                Long sourceDsId = syncTaskByNodeId.getSourceDsId();
                Long targetDsId = syncTaskByNodeId.getTargetDsId();
                DataSourceDetailDomain dataSourceDetailById =
                    dataSourceService.getDataSourceDetailById(sourceDsId, false);
                DataSourceDetailDomain targetSourceId =
                    dataSourceService.getDataSourceDetailById(targetDsId, false);
                if (dataSourceDetailById.getIsSystem().equals(1)) {
                    res.add(syncTaskByNodeId.getSourceTable());
                } else if (targetSourceId.getIsSystem().equals(1)) {
                    res.add(syncTaskByNodeId.getTargetTable());
                }
            } else if (nodeType.equals(OfflineDevNodeTypeEnum.API.getNodeType())) {
                OfflineDevApiDomain apiTaskByNodeId = datadevOfflineApiService.getApiTaskByNodeId(id);
                TaskConfigDomain taskConfigDomain = apiTaskByNodeId.getTaskConfigDomain();
                String targetTableName = taskConfigDomain.getTargetTableName();
                res.add(targetTableName);
            }
        }
        return Lists.newArrayList(res);
    }
}
